Interview prep — coding drills — Agentic AI Engineering

These are the live-coding tasks you will get on a 60-minute pair-programming round. Practise each at least once without internet, then once with internet (real-world conditions). Speed matters — but more important is narrating decisions as you code.

Drill 1 — "Build a structured-output JSON extractor (no framework)"

Brief: extract {name, email, phone, address} from messy customer messages. Return a Pydantic model. Auto-retry on validation failure.

Skeleton you should write in 15 minutes:

python

from openai import OpenAI
from pydantic import BaseModel, EmailStr, ValidationError, Field
import json

class Contact(BaseModel):
    name: str
    email: EmailStr
    phone: str = Field(..., pattern=r"^\+?[0-9 \-]+$")
    address: str | None = None

client = OpenAI()

def extract(text: str, max_retries: int = 3) -> Contact:
    last_error = ""
    for _ in range(max_retries):
        prompt = f"Extract a contact JSON.\nERROR (if any): {last_error}\nINPUT: {text}"
        r = client.responses.parse(
            model="gpt-4.1-mini",
            instructions="Return only the JSON.",
            input=prompt,
            text_format=Contact,
        )
        try:
            return r.output_parsed
        except ValidationError as e:
            last_error = str(e)
    raise RuntimeError(f"failed after retries: {last_error}")

Talking points: strict schema, retry-with-error-message, why Pydantic.

Drill 2 — "Manual ReAct loop"

Brief: build a 50-line agent loop with two tools (get_weather, convert_currency). No LangChain.

python

import json
from openai import OpenAI
from pydantic import BaseModel, Field

class WeatherArgs(BaseModel):
    city: str

class FxArgs(BaseModel):
    amount: float = Field(..., gt=0)
    from_: str
    to_: str

def get_weather(args: WeatherArgs): return {"city": args.city, "tempC": 32}
def convert(args: FxArgs):          return {"converted": args.amount * 84.0}

TOOLS = {
    "get_weather": (WeatherArgs, get_weather),
    "convert_currency": (FxArgs, convert),
}

def to_schema(name, A):
    return {"type":"function","function":{"name":name,
            "description":f"{name} tool",
            "parameters":A.model_json_schema(),"strict":True}}

client = OpenAI()
def run(user_msg, max_steps=8):
    history = [{"role":"system","content":"Use tools when helpful."},
               {"role":"user","content":user_msg}]
    for _ in range(max_steps):
        r = client.chat.completions.create(
            model="gpt-4.1-mini", messages=history,
            tools=[to_schema(n,A) for n,(A,_) in TOOLS.items()],
            tool_choice="auto",
        )
        m = r.choices[0].message
        history.append(m.model_dump(exclude_unset=True))
        if not m.tool_calls: return m.content
        for c in m.tool_calls:
            A, fn = TOOLS[c.function.name]
            args = A(**json.loads(c.function.arguments))
            try:
                out = fn(args)
            except Exception as e:
                out = {"error": str(e)[:200]}
            history.append({"role":"tool","tool_call_id":c.id,"content":json.dumps(out)})
    return "out of steps"

Talking points: strict schema, registry pattern, error swallowing, step cap.

Drill 3 — "Hybrid retriever + RRF + rerank"

Brief: given a corpus and a query, return top-5 with hybrid + Cohere rerank-3.

python

import bm25s, chromadb, cohere
from collections import defaultdict

co = cohere.ClientV2()

def index(corpus):
    cli = chromadb.Client(); col = cli.get_or_create_collection("c")
    col.add(ids=[str(i) for i in range(len(corpus))], documents=corpus)
    bm = bm25s.BM25(); bm.index(bm25s.tokenize(corpus))
    return col, bm

def rrf(lists, k=60):
    s = defaultdict(float)
    for L in lists:
        for r, x in enumerate(L, 1): s[x] += 1.0/(k+r)
    return [x for x,_ in sorted(s.items(), key=lambda kv: -kv[1])]

def retrieve(q, col, bm, corpus, k=5):
    dense = col.query(query_texts=[q], n_results=50)["ids"][0]
    bm_top, _ = bm.retrieve(bm25s.tokenize([q]), k=50)
    bm_ids = [str(i) for i in bm_top[0]]
    fused = rrf([dense, bm_ids])[:50]
    docs  = [corpus[int(i)] for i in fused]
    rr = co.rerank(model="rerank-3", query=q, documents=docs, top_n=k)
    return [docs[r.index] for r in rr.results]

Talking points: dual store IDs, RRF k=60, cap rerank at 50.

Drill 4 — "LangGraph 3-node graph with HIL"

Brief: agent that drafts an email, asks a human to approve, then "sends" it.

python

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import InMemorySaver
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4.1-mini")

class S(TypedDict):
    topic: str
    draft: str
    approved: bool

def draft(s):
    return {"draft": llm.invoke(f"write a 4-sentence email about {s['topic']}").content}

def review(s):
    out = interrupt({"draft": s["draft"], "ask": "approve?"})
    return {"approved": bool(out)}

def send(s):
    if not s["approved"]: return {"draft": s["draft"] + "\n[CANCELLED]"}
    print("sending:", s["draft"])
    return {}

g = StateGraph(S)
g.add_node("draft", draft); g.add_node("review", review); g.add_node("send", send)
g.add_edge(START,"draft"); g.add_edge("draft","review"); g.add_edge("review","send"); g.add_edge("send", END)
graph = g.compile(checkpointer=InMemorySaver())

cfg = {"configurable": {"thread_id":"t1"}}
print(graph.invoke({"topic":"team off-site"}, config=cfg))   # pauses
print(graph.invoke(Command(resume=True), config=cfg))         # resumes

Talking points: checkpointer required for interrupt; resume re-runs node from start, so side-effects after interrupt.

Drill 5 — "Cost-aware cascade"

Brief: build cascade(question) that tries gpt-4.1-mini with confidence, escalates to gpt-5.5 only when confidence < 0.7. Log cost.

python

from openai import OpenAI
from pydantic import BaseModel
client = OpenAI()

class Ans(BaseModel):
    answer: str
    confidence: float

def call(model, q):
    r = client.responses.parse(model=model, input=q, text_format=Ans)
    usage = r.usage
    cost = usage.input_tokens / 1_000_000 * (0.40 if "mini" in model else 5.00) \
         + usage.output_tokens / 1_000_000 * (1.60 if "mini" in model else 20.00)
    return r.output_parsed, cost

def cascade(q):
    a, c1 = call("gpt-4.1-mini", q)
    if a.confidence >= 0.7:
        return a.answer, c1, "mini"
    b, c2 = call("gpt-5.5", q)
    return b.answer, c1+c2, "cascade"

Talking points: Pydantic confidence is self-rated; combine with heuristics in real prod.

Drill 6 — "FastMCP server in 30 lines"

Brief: ship a small MCP server with one tool and one resource.

python

from fastmcp import FastMCP
mcp = FastMCP(name="DemoMCP")

@mcp.tool
def add(a: int, b: int) -> int:
    """Add two integers."""
    return a + b

@mcp.resource("config://app")
def get_config() -> dict:
    """App config."""
    return {"version":"1.0", "owner":"demo"}

if __name__ == "__main__":
    mcp.run()

Talking points: stdio vs HTTP transports, Inspector for testing, integration with Claude Desktop.

Drill 7 — "Trace decorator + provider-agnostic call"

Brief: decorate a function so its inputs/outputs/timings appear in LangSmith. Provider-agnostic via litellm.

python

import time
from langsmith import traceable
from litellm import completion

@traceable(run_type="llm", name="ask")
def ask(prompt: str, model="anthropic/claude-haiku-4-5") -> str:
    t = time.time()
    r = completion(model=model, messages=[{"role":"user","content":prompt}])
    dt = time.time() - t
    return r.choices[0].message.content + f"\n[latency: {dt:.2f}s]"

Set LANGSMITH_TRACING=true and LANGSMITH_API_KEY=.... The trace appears in LangSmith automatically.

Drill 8 — "Eval gate"

Brief: write a pytest test that asserts an agent's faithfulness on 20 cases is ≥ 0.85.

python

def test_faithfulness_at_least_085(my_rag, golden):
    from ragas import EvaluationDataset, evaluate
    from ragas.metrics import Faithfulness
    samples = [
        {"user_input": x["q"], "response": my_rag(x["q"]),
         "retrieved_contexts": [c["text"] for c in retrieve(x["q"])],
         "reference": x["a"]}
        for x in golden[:20]
    ]
    res = evaluate(EvaluationDataset.from_list(samples), metrics=[Faithfulness()])
    assert res["faithfulness"] >= 0.85, res

Run in CI; PR with regressions fails.

Drill 9 — "Reduce a runaway agent's cost by 50%"

Asked verbally. Your answer should hit:

Quantify current cost with logs.
Add prompt caching (Anthropic 90% / OpenAI auto).
Cascade cheap → expensive.
Trim history; summarise older turns.
Cap max_tokens.
Move to batch API for non-realtime.
Self-host with vLLM if volume warrants.

Bonus: mention you would measure each change and roll back if quality drops.

Tips for live coding rounds

Narrate every choice ("I'm using responses.parse because the SDK auto-validates").
Type-hint everything.
Wrap external calls in try/except.
Add a 3-line docstring to every function.
When stuck, say "I would normally look this up — here is what I think the answer is."

Practise these 9 drills until you can finish each in under 25 minutes.