AI Engineer Interview Experience

Write a Python function to process and clean a dataset.

import pandas as pd def clean_dataset(df): df = df.copy() # Remove duplicate rows df = df.drop_duplicates() # Remove completely empty rows df = df.dropna(how="all") # Normalize column names df.columns = ( df.columns .str.strip() .str.lower() .str.replace(" ", "_") ) # Clean string columns for column in df.select_dtypes(include="object").columns: df[column] = df[column].str.strip() return df

I would use least-privilege tool permissions, strict input validation, authentication and authorization checks, and allowlisted actions. For sensitive operations, I would require human confirmation before execution. I would also validate the agent's proposed tool arguments instead of trusting the model output directly. On top of that, I would add rate limits, execution limits, audit logging, timeouts, and monitoring so an agent cannot repeatedly perform an action or operate outside its intended scope.

I would first determine whether the operation is idempotent. For operations that create or modify data, I would use an idempotency key or unique request identifier so the server can recognize repeated requests and return the original result instead of performing the operation twice. I would combine that with limited retries, exponential backoff, jitter, and clear timeout handling. I would only retry errors that are actually transient, such as timeouts, 429 responses, or certain 5xx errors.

A list is ordered and mutable, so I use it when I need a collection that can change and where order matters. A tuple is also ordered, but it's immutable, so it's useful for values that shouldn't change. A set stores unique values and is useful for removing duplicates or doing fast membership checks. A dictionary stores key-value pairs and is what I normally use when I need to associate one piece of information with another.

import asyncio import httpx async def fetch(client, url): response = await client.get(url, timeout=10.0) response.raise_for_status() return response.json() async def fetch_all(urls): async with httpx.AsyncClient() as client: tasks = [fetch(client, url) for url in urls] results = await asyncio.gather( *tasks, return_exceptions=True ) return results

I start with what I can verify instead of immediately trusting the error message. I reproduce the request if possible and trace it through the endpoint, service layer, database, and any external APIs. I compare timestamps, request IDs, status codes, payloads, and the last successful execution. If the logs are incomplete, I'll add targeted instrumentation around the boundary where the behavior becomes inconsistent. I try to change one thing at a time so I can prove which component is actually responsible instead of fixing several possible causes at once.

I think of RAG as two main pipelines: ingestion and retrieval/generation. During ingestion, I load the source documents, clean and split them into meaningful chunks, attach metadata, generate embeddings, and store those embeddings in a vector database. When a user submits a question, I embed the query and retrieve the most relevant chunks. Depending on the application, I may apply metadata filtering, hybrid search, or reranking. I then provide the best retrieved context to the LLM with instructions to answer from that evidence. I also evaluate retrieval and generation separately because a bad answer can come from either poor retrieval or poor generation.

One system I worked on was a document question-answering application using RAG. I handled document ingestion and preprocessing, split the content into chunks, generated embeddings, and stored them in a vector database. I then built the retrieval layer to find relevant context for each question and connected that context to an LLM through a backend API. After getting the basic pipeline working, I created test questions and evaluated retrieval relevance, groundedness, and answer quality. I used the failures to adjust chunking, retrieval parameters, prompts, and error handling rather than relying on a few successful examples.

I wouldn't rely on the model itself to decide when it has run too long. I would enforce a maximum number of steps or tool calls at the application level and set execution timeouts. I would also track recent actions so the system can detect repeated tool calls with the same arguments or repeated states without meaningful progress. If that happens, I would stop execution and either return a controlled failure or escalate for human review. For higher-risk agents, I would also use explicit state transitions so the model can't freely loop between tools.

I would treat that as a system-level control rather than relying only on the prompt. Before anything is sent to the external model, I would run the input through a detection and redaction layer for information such as names, email addresses, phone numbers, account identifiers, and other sensitive fields. Where possible, I would replace those values with placeholders or internal IDs and restore them only after processing when necessary. I would also minimize what data is sent, restrict logging of sensitive payloads, encrypt data in transit, control access, and make sure the model provider and data-retention configuration meet the application's privacy requirements.

I focus on the impact of the decision instead of overwhelming people with implementation details. For example, instead of only saying that one model has lower latency, I would explain that it gives users faster responses and costs less to operate, but we may give up some accuracy on more difficult requests. I usually present the available options, the benefits and risks of each, my recommendation, and why I'm recommending it. That gives stakeholders enough information to make a decision while still keeping the technical reasoning clear.