Python Interview Questions for AI Engineers

Approach: Explain that Python passes object references, not copies — so a mutable list shared between caller and function is the same object. Then describe defensive copying, immutability, and documentation as prevention strategies.

• Python's argument passing is 'pass-by-object-reference' — the function receives a reference to the same list object, not a copy.
• list.append() mutates the object in place, so the caller's reference now sees the new item.
• In a pipeline component that processes data for multiple downstream consumers, unintended mutation can cause data corruption, race conditions, or silent bugs.
• Prevention options: (a) copy.deepcopy for nested structures, (b) return a new list instead of mutating in place, (c) use tuple or frozen dataclass for data that should not change, (d) document mutation behaviour explicitly in the function contract.

Trade-offs:

• Defensive copying is safe but costs memory and time — unacceptable for large ML tensors or big document collections
• Immutability via tuples or frozen dataclasses prevents mutation but makes transformations more verbose (create new objects instead of modifying)
• Documentation is zero-cost but relies on caller discipline — fine for small teams, risky for shared libraries

Failure modes:

• Shallow copy (list.copy()) does not protect nested mutable objects — a dict inside a copied list is still shared
• Using tuple() on a list of dicts makes the outer container immutable but the dicts inside are still mutable
• Assuming 'pass-by-value' semantics — this is a fundamental Python misconception

Validation:

• Write a test that checks the original list is unchanged after calling the function
• Use id() to demonstrate that the function's parameter and the caller's variable point to the same object
• Show that copy.deepcopy produces a different id()

def safe_process(items: list[dict]) -> list[dict]:
    # Return new list — do not mutate caller's data
    return [{**item, "processed": True} for item in items]

# Caller's list is untouched
original = [{"id": 1}, {"id": 2}]
result = safe_process(original)
assert original[0] == {"id": 1}  # No mutation

Approach: Explain that generators produce values lazily, one at a time, so they avoid holding all 500K documents in memory simultaneously. Then identify the trade-off: generators cannot be indexed, replayed, or parallelised without materialisation.

• A list of 500K document dicts could consume gigabytes of RAM — a generator yielding one document at a time uses constant memory.
• Generators are ideal for streaming pipelines: read document → chunk → embed → store → yield next.
• However, generators are single-pass — once consumed, they are exhausted. If you need to retry failed documents or run multiple passes (e.g., re-embed after a model change), you need a list or a re-creatable generator.
• Generators cannot be sliced, indexed, or passed to multiprocessing.Pool.map directly without being converted to a list first.

Trade-offs:

• Memory: generator wins (O(1) vs O(n) memory)
• Replayability: list wins (can iterate multiple times, slice, index)
• Parallelism: list wins (can chunk for multiprocessing; a generator must be materialised first)
• Error recovery: list wins (can retry specific indices; a generator that raises mid-iteration loses position)

Failure modes:

• Using a generator for a pipeline that needs retry — when an exception occurs at item 300K, the generator is exhausted and you cannot resume
• Passing a generator to multiprocessing.Pool.map — it materialises the entire generator into a list internally, defeating the memory saving
• Assuming a generator is always faster — for small datasets, the overhead of generator creation and next() calls can be slower than a simple list comprehension

Validation:

• Measure peak memory with sys.getsizeof or tracemalloc for list vs generator on a sample dataset
• Time both approaches — for small datasets, list comprehension may be faster
• Test error recovery: raise an exception mid-iteration and verify whether the generator can resume

Approach: Define a Protocol with chat() and stream_chat() methods, explain that Protocol is structural typing (duck typing with static checks) while ABC is nominal typing (explicit inheritance), and state when each fits.

• Protocol (PEP 544) enables structural subtyping — any class that has the right methods satisfies the Protocol, without inheriting from it.
• This is ideal for LLM providers: OpenAI's SDK, Anthropic's SDK, and a custom Ollama wrapper do not share a base class, but they can all implement the same method signatures.
• ABC (abstract base class) requires explicit inheritance — the provider class must inherit from the ABC. This is better when you control all implementations and want to enforce a contract at runtime.
• For a provider-agnostic AI service, Protocol is better because you cannot make OpenAI's SDK inherit from your ABC.

Trade-offs:

• Protocol: no runtime enforcement by default (use @runtime_checkable for isinstance checks, but it only checks method existence, not signatures)
• ABC: runtime enforcement via abstractmethod, but requires inheritance from your base class
• Protocol: works with third-party classes you cannot modify
• ABC: better for internal hierarchies where you want shared logic in the base class

Failure modes:

• Using @runtime_checkable and assuming it validates method signatures — it only checks method names exist, not parameter types
• Making the Protocol too narrow (only OpenAI's method shape) so Anthropic cannot satisfy it without adapters
• Forgetting that Protocol classes are not instantiated — they are type hints, not base classes

Validation:

• Run mypy or pyright with strict mode to verify that each provider class satisfies the Protocol
• Write a test that calls chat() on each provider through a function typed to accept the Protocol
• Verify that removing a required method from a provider causes a type error

from typing import Protocol, Iterator

class LLMProvider(Protocol):
    def chat(self, messages: list[dict], **kwargs) -> str: ...
    def stream_chat(self, messages: list[dict], **kwargs) -> Iterator[str]: ...
    def structured_output(self, messages: list[dict], schema: type, **kwargs) -> dict: ...

# Any class with these methods satisfies LLMProvider
class OpenAIWrapper:
    def chat(self, messages, **kwargs) -> str: ...
    def stream_chat(self, messages, **kwargs) -> Iterator[str]: ...
    def structured_output(self, messages, schema, **kwargs) -> dict: ...

def call_provider(p: LLMProvider, msgs: list[dict]) -> str:
    return p.chat(msgs)  # mypy verifies p has chat()

Approach: Use a context manager (with statement) for the database connection so it is closed on any exit path. Wrap the LLM call in a try/except for timeout, log the failure, and let the context manager handle cleanup.

• The 'with' statement guarantees __exit__ is called on the connection even if an exception propagates — this is the standard Python resource safety pattern.
• If the LLM call raises a TimeoutError, the exception propagates up through the 'with' block, which triggers the connection's __exit__ (close) before the exception continues.
• For async services (FastAPI), use 'async with' for async database drivers (asyncpg, databases) — the same guarantee applies.
• If you need to clean up multiple resources (connection + HTTP client + file), nest context managers or use ExitStack for dynamic resource management.
• Cancellation (asyncio.CancelledError) is distinct from synchronous TimeoutError. In async code, a cancelled task propagates CancelledError through the async with block, which still triggers __aexit__. The cleanup path is the same, but you should not catch CancelledError as if it were a timeout — let it propagate so the caller knows the task was cancelled, not timed out.
• Never log raw prompts or records that may contain PII. Log only metadata (prompt length, model name, latency, status code) or use a redaction filter before logging.

Trade-offs:

• Nesting context managers creates indentation — use contextlib.ExitStack or async-exit-stack for many resources
• Catching the timeout inside the with-block lets you log and return a fallback, but masks the error from upstream — decide based on whether the caller can handle it
• For connection pooling (SQLAlchemy, asyncpg pool), 'with' returns a connection to the pool rather than closing it — the pool itself manages lifecycle
• try/finally is a valid alternative that guarantees cleanup. Context managers provide a cleaner abstraction because they encapsulate the acquire/release pair, reduce boilerplate, and are harder to get wrong with multiple return paths.

Failure modes:

• Opening the connection outside the with-block and closing it manually — if the LLM call raises, the close() line is never reached
• Using try/finally instead of a context manager — correct but more verbose and error-prone (forgetting finally on one of multiple return paths)
• In async code, using 'with' instead of 'async with' for an async resource — the resource is never properly closed
• Catching asyncio.CancelledError and treating it as a timeout — this prevents proper task cancellation propagation and can leave the caller waiting indefinitely.

Validation:

• Test with a mock LLM that raises TimeoutError and verify the connection is closed using a connection pool metric
• Use tracemalloc or a connection counter to confirm no connection leak after 1000 iterations with random failures
• In async tests, use pytest-asyncio and assert the pool returns to its initial connection count

def process_with_llm(db_pool, llm_client, prompt: str) -> str:
    with db_pool.get_conn() as conn:  # Closed on any exit
        try:
            result = llm_client.complete(prompt, timeout=30)
        except TimeoutError:
            log.warning("LLM timeout for prompt: %s", prompt[:100])
            return "[timeout]"
        conn.execute(
            "INSERT INTO results (prompt, output) VALUES (?, ?)",
            prompt, result
        )
        return result

Approach: Use dependency injection to separate the LLM call from the business logic, mock the LLM client in tests with deterministic fixtures, and test the real API integration separately with a smoke test that is skipped in CI.

• The service should accept an LLM client as a parameter (dependency injection), not create it internally — this allows tests to inject a mock.
• Unit tests mock the LLM client to return fixed responses — they test the service's logic (prompt construction, response parsing, error handling, database writes) without calling the API.
• The mock should return realistic response shapes (not just 'hello') so parsing logic is tested — use fixtures stored as JSON files.
• For the real API integration, write a separate smoke test marked with @pytest.mark.integration that is skipped in CI via a marker or environment variable.

Trade-offs:

• Mock tests are fast and deterministic but do not catch API changes (e.g., response format changes, new error types)
• Recording real API responses (VCR.py) captures actual formats but recordings go stale when the API changes
• Calling the real API in tests gives the most realistic coverage but is slow, costly, and nondeterministic — flaky tests erode trust

Failure modes:

• Mocking at too high a level — mocking the entire service function rather than just the LLM client, so the test does not exercise internal logic
• Using a fixed seed or temperature=0 in tests — this reduces randomness but is still nondeterministic across model versions
• Storing API keys in test files or CI environment without rotation — a security risk

Validation:

• Run unit tests 100 times in CI and confirm zero flaky failures
• Periodically run the integration smoke test manually against the real API to catch format changes
• Use a contract test that validates the mock response shape against the real API response shape

Approach: Use asyncio for the embedding API call (I/O-bound — awaiting network response) and a process pool or separate worker for the reranking model (CPU-bound — asyncio does not help with CPU work). Explain that async does not increase CPU parallelism.

• asyncio uses a single-threaded event loop — it improves I/O concurrency by switching tasks while one waits for network, but it cannot run CPU-bound work in parallel.
• The embedding API call is I/O-bound: the coroutine awaits an HTTP response, freeing the event loop to handle other requests. asyncio is correct here.
• The reranking model is CPU-bound: running it inside the event loop blocks all other requests until it finishes. This is the classic 'blocking the event loop' problem.
• Solution: run the reranking model in a process pool (concurrent.futures.ProcessPoolExecutor or multiprocessing) and await its result from the async context using loop.run_in_executor().
• Generators can be parallelised with careful design (e.g., chunking into batches and submitting each batch to a pool), but a single-pass generator cannot be split across workers without materialising it first.

Trade-offs:

• Process pool: true CPU parallelism but process creation overhead and data serialisation cost (pickle the model input/output)
• Thread pool: no serialisation cost but Python's GIL means only one thread runs Python at a time — useful if the reranker releases the GIL (e.g., numpy/PyTorch C extensions)
• GPU inference: if the reranker runs on GPU, use a dedicated inference server (vLLM, TGI) and call it via async HTTP — the GPU handles parallelism, and the async call is I/O-bound from Python's perspective

Failure modes:

• Running a CPU-bound reranker inside the event loop — all concurrent requests stall until the reranker finishes, causing p99 latency spikes
• Using threading for a pure-Python CPU workload — the GIL prevents true parallelism, so threads do not help
• Forgetting that loop.run_in_executor() with a process pool pickles arguments — large tensors or complex objects may fail to pickle

Validation:

• Benchmark: send 10 concurrent requests and measure p99 latency — if the reranker blocks the event loop, p99 will be ~10x the single-request latency
• After moving to a process pool, p99 should drop to ~1-2x the single-request latency (depending on pool size)
• Use asyncio.get_event_loop().run_in_executor(None, fn, args) for a quick test — None uses the default thread pool, then switch to ProcessPoolExecutor for CPU-bound work

# Create pool ONCE at app startup, not per request
_executor = ProcessPoolExecutor(max_workers=4)

async def rag_handler(query, embed_client, reranker):
    embeddings = await embed_client.embed(query)  # I/O-bound: async
    loop = asyncio.get_event_loop()
    ranked = await loop.run_in_executor(_executor, reranker.rerank, embeddings)
    return ranked

Approach: Use structured timing around each pipeline stage (not just the total) to measure where time is actually spent. Profile with cProfile for CPU time and identify whether the bottleneck is embedding, retrieval, LLM generation, or serialisation.

• Start with stage-level timing: wrap each pipeline stage (embed, retrieve, rerank, generate) with time.perf_counter() and log the duration. This gives a breakdown without profiler overhead.
• If stage timing shows embedding is 0.2s and LLM generation is 3.2s, the hypothesis is refuted — the bottleneck is the LLM, not the embedding model.
• Use cProfile for deeper analysis if stage timing is insufficient — it shows per-function call counts and cumulative time, revealing hidden costs (e.g., JSON serialisation, regex parsing, database connection overhead).
• For async services, use asyncio debug mode and tracing (e.g., OpenTelemetry spans) to measure time spent awaiting each I/O operation.

Trade-offs:

• Stage-level timing: minimal overhead, easy to add, but does not show intra-stage breakdowns
• cProfile: detailed per-function data but adds 10-30% overhead and is not suitable for production traffic
• py-spy (sampling profiler): low overhead, works on production, but less detailed than cProfile
• OpenTelemetry tracing: production-safe, distributed, but requires instrumentation setup

Failure modes:

• Optimising a component without profiling first — if it is a small fraction of total time, even a large improvement saves little.
• Using time.time() instead of time.perf_counter() — time.time() can go backwards on NTP adjustments and has lower resolution
• Profiling in development with warm caches and concluding the bottleneck is small — the production bottleneck may be different (cold cache, network latency, database load)

Validation:

• Run the profiler on production-like data and traffic patterns, not just a single test query
• Measure p50 and p95 separately — the p95 bottleneck may differ from the p50 bottleneck
• After optimisation, re-profile to confirm the bottleneck moved — do not assume the fix worked without measurement

Approach: Move from symptom to instrumented isolation: log the failing records, reproduce locally with the same inputs, inspect the transformation at each step, identify whether the issue is in parsing, type coercion, or missing-key handling, and write a regression test before fixing.

• Step 1 — Capture: Log the failing records with their input JSON and output. This tells you whether the input is malformed or the transformation is wrong.
• Step 2 — Reproduce: Take one failing input and run the transformation locally. Inspect the output at each step: parse, validate, transform, serialize.
• Step 3 — Isolate: If the input JSON has a field that is null instead of a string, the issue is in the API response. If the input is correct but the output has missing fields, the issue is in the transformation logic — a dict.get() that silently returns None for unexpected keys.
• Step 4 — Fix: The 2% failure rate suggests an edge case — the API occasionally returns a field as null or a different type. Fix the transformation to handle these cases explicitly with type checking, default values, or rejection with logging.
• Step 5 — Regression test: Write a test with the failing input shape (null field, missing key, wrong type) and verify the transformation handles it correctly.

Trade-offs:

• Silent defaults (dict.get(key, default)) hide data quality issues — the pipeline appears to work but downstream ML features get incorrect values
• Strict validation (raising on unexpected types) catches issues early but may reject valid-but-unusual API responses
• Logging all transformations is expensive — sample or log only failures

Failure modes:

• Assuming the API always returns the documented schema — real APIs have bugs, version skew, and edge cases
• Using dict.get() without checking the result — None propagates silently through the pipeline
• Not logging the original input alongside the output — without both, you cannot reproduce the failure

Validation:

• Run the regression test with the failing input shape and confirm it passes
• Add a schema validation step (pydantic or jsonschema) at the pipeline entry to catch malformed inputs early
• Monitor the failure rate after the fix and confirm it drops to zero

# Bad: silent None propagation
record = {"feature": data.get("value")}  # None if missing

# Good: explicit handling
value = data.get("value")
if value is None:
    log.warning("Missing value in record %s", record_id)
    continue
record = {"feature": float(value)}