Full-Stack AI interview questions & answers

Picture a home kitchen versus a restaurant kitchen. At home, one person cooks for whoever is standing in the room, using whatever is in the fridge, and if the stove breaks dinner is just late tonight. A restaurant kitchen cooks for forty tables at once, has to know exactly which dish belongs to which table, tracks what each ingredient costs, and has a plan for the night the gas line fails mid-service. Same stove, same recipes -- completely different job.

Every chapter before this one was the home kitchen. RAG (Ch4), tool calling (Ch5), LangChain agents (Ch6), memory (Ch7), fine-tuning (Ch8), guardrails (Ch9) -- all of it ran as one user, one machine, one request at a time, with an API key sitting quietly in a variable and a print() at the end. That is a script, and a script is allowed to ignore four problems because there is only ever one person in the room to be affected by them.

The moment a second user shows up, those four problems stop being optional:

• Concurrency -- many requests arrive close together, sometimes at the exact same instant. A script processes one thing after another; a product has to decide what happens to request two while request one is still running.
• Identity -- "whose data is this?" A script's answer was always "mine" and never had to be asked. A product must know, for every single request, which user is asking -- because their chat history, their documents, their usage, and (as later topics in this chapter show) even their cached answers all depend on getting that right.
• Cost -- someone is paying per token, on every call, from every user, all day. A script is one experiment; a product is a running bill that keeps a tab even while you sleep.
• Failure -- the model provider is a remote service across the internet. It will time out, return an error, or fail midway through a response, and it will not politely wait for a convenient moment to do it.

None of those four are about the prompt. You can have the best prompt in the world and still ship a backend that freezes for every user the moment one slow request comes in, or leaks one customer's chat history into another customer's screen. This chapter is about the plumbing around the model call -- the part a notebook never needed and a product cannot ship without.

To keep that plumbing straight, hold one diagram in your head for the rest of this chapter -- the request path:

browser -> auth -> quota -> cache -> [ RAG / tools / memory ] -> provider -> stream back -> log

Every later topic in this chapter is one station on that path: fastapi-basics and async-vs-sync-endpoints are how the server accepts the request at all; auth-multitenancy is the auth box; usage-metering-quotas is the quota box; caching-llm-responses is the cache box; streaming-sse-backend / vercel-ai-sdk / streaming-ui-frontend are the stream back box; error-handling-timeouts covers what happens when the provider box misbehaves; background-jobs-queues is for work that should never sit on this path at all; and ai-backend-architecture draws the whole picture with all the stations in place. The [ RAG / tools / memory ] box in the middle is not new work -- it is exactly Ch4, Ch5, Ch6 and Ch7, called as a function from inside this path. This chapter does not re-teach what happens inside that box; it teaches everything around it.

The snippet below makes the smallest possible version of that jump concrete: the same fake model call, first run as a bare script (exactly like every earlier chapter), then run again behind one FastAPI endpoint using TestClient. Watch closely for what changes and what does not -- the honest answer is "less than you'd think," and that gap is this chapter's real subject.

🌍 Real-world example: a college project that summarizes one PDF for one user, run from a Jupyter cell, is a script. The moment you deploy it so classmates can each upload their own PDF and get their own summary back at the same time, you owe it concurrency (don't let one big PDF freeze everyone else's request), identity (whose PDF and whose summary is this), cost (someone's account is being billed per upload), and failure handling (what does a classmate see if the model call times out) -- none of which existed as a problem when it was just you, in one cell, running it once.

💡 Script = code that serves one user, one request at a time, with no separation between "my call" and "someone else's call" because there is no one else.

💡 Product = the same core logic, now reachable by many users at once, each of whom needs their own identity, their own cost tracked, and a defined answer for what they see when something fails.

💡 The request path = the fixed sequence every request travels: browser -> auth -> quota -> cache -> RAG/tools/memory -> provider -> stream back -> log. Every topic in this chapter is one station on it.

When to use it: treat a project as "just a script" only while exactly one person will ever call it at a time -- a personal tool, a one-off data job, a demo you run yourself. The moment a second, independent caller exists (a teammate, a classmate, a paying user), you are building a product whether you intended to or not, and the four problems above are already live.

When NOT to over-engineer this: you do not need auth, a cache layer, and a job queue on day one. The honest early architecture is one process that already separates "who is asking" from "what they asked," so that adding real auth or a real cache later is a small change, not a rewrite. Building three services before you have a single real second user is its own mistake -- this chapter's later ai-backend-architecture topic comes back to exactly this trade-off.

Standard definition: A script and a product can share the same model call and the same prompt; what separates them is everything that surrounds that call -- concurrency (serving many requests at once instead of one), identity (knowing which user a request belongs to), cost (someone paying per token across every call, not just yours), and failure handling (the provider is a remote service that will time out or error, and the system must have a defined behaviour for that moment) -- collectively "the request path," which this chapter builds one station at a time.

from fastapi import FastAPI
from fastapi.testclient import TestClient

# The fake "LLM" -- no API key, no network call. Every earlier chapter's
# script called something exactly this shape and printed the result directly.
def fake_llm(prompt: str) -> str:
    return f"echo-model: you said '{prompt}'"

# ---- Step 1: the notebook / script version -----------------------------
# One user (you), one request, a key sitting in a variable (there isn't
# even one here, because there is nothing else running to protect it from).
prompt = "What is a REST API?"
answer = fake_llm(prompt)
print("[script]", answer)

# ---- Step 2: the same call, behind one FastAPI endpoint -----------------
app = FastAPI()

@app.post("/chat")
def chat(payload: dict):
    return {"answer": fake_llm(payload["prompt"])}

client = TestClient(app)
resp = client.post("/chat", json={"prompt": "What is a REST API?"})
print("[api]   status:", resp.status_code, "body:", resp.json())

# ---- What the wrapper added, and what it did NOT ------------------------
print()
print("added by @app.post:      an HTTP address other machines can call")
print("added by @app.post:      request/response parsing into JSON")
print("still missing:           only one process -- two users calling at")
print("                         the same instant just queue up")
print("still missing:           no auth -- payload has no notion of WHO is asking")
print("still missing:           no cost control -- nothing stops 1 call or 1 million")

Think of a restaurant order counter. In Express, you built that counter yourself: you wrote the sign that lists what a valid order looks like, you personally checked whether the customer actually filled in every field on the order slip, and if a customer wrote "five" instead of a number in the quantity box, your own code had to notice and reject it. FastAPI hires a counter clerk who already knows the order form by heart: you hand the clerk a description of what a valid order slip looks like, and the clerk checks every order against it before it ever reaches you -- no notice-and-reject code of your own required.

In FastAPI, path/query/body handling maps directly onto what you already know from Express -- same three inputs, different plumbing:

Express	FastAPI
app.get('/items/:id', (req, res) => { req.params.id })	@app.get("/items/{item_id}") with def read_item(item_id: int)
req.query.q (always a string, you convert it)	q: str | None = None as a function parameter -- typed and optional by default value
req.body, then express-validator or a hand-rolled if (!req.body.message)	a pydantic request model: class ChatRequest(BaseModel): message: str
npm install swagger-jsdoc swagger-ui-express + writing YAML comments above every route	/docs and /openapi.json exist automatically, generated from the same code you already wrote
node server.js / nodemon	uvicorn main:app --reload -- the ASGI server that actually runs your FastAPI app; FastAPI is the framework, uvicorn is what executes it, the same way Express is the framework and Node is what runs it

What is genuinely different, not just renamed: in Express, validation and docs are two separate libraries you bolt on -- express-validator for one, swagger-jsdoc for the other -- and both can silently drift out of sync with your actual route code, because nothing forces them to agree. In FastAPI, the pydantic model IS the validation, and the same model IS the documentation. You declare the shape once; FastAPI generates both from that single source of truth. There is no separate schema file to forget to update.

Here is the request model doing the checking for you:

class ChatRequest(BaseModel):
    message: str
    temperature: float = Field(default=0.7, ge=0.0, le=2.0)

Send a request missing message, and FastAPI never even calls your function -- it rejects the request itself:

422 {'detail': [{'type': 'missing', 'loc': ['body', 'message'], 'msg': 'Field required', 'input': {'temperature': 0.5}}]}

Send temperature: 5 against a field declared ge=0.0, le=2.0, and the same thing happens:

422 {'type': 'less_than_equal', 'loc': ['body', 'temperature'], 'msg': 'Input should be less than or equal to 2', 'input': 5, 'ctx': {'le': 2.0}}

Three things to notice, and they are exactly the ones that trip up an Express developer moving over. First, it is 422 Unprocessable Entity, not 400 Bad Request -- 400 means the request itself is malformed (broken JSON); 422 means the JSON parsed fine but did not satisfy the declared shape. If your frontend only checks if (status === 400) it will silently miss every validation failure. Second, the loc array tells you exactly which field failed -- ['body', 'temperature'] -- which is precisely what a frontend needs to highlight the right input box red, and you did not write a single line of code to produce it. Third, this happens for every endpoint using this model, automatically -- there is no if not request.body.message: return 400 anywhere in the handler above. Declare the shape once; the checking is free forever after.

Now the other direction: response_model is not documentation, it is a filter on the way OUT. Declare what a response is supposed to contain, and FastAPI strips anything the handler returns that is not in that shape -- even if the handler itself puts it there by mistake:

class ChatResponse(BaseModel):
    answer: str
    tokens_used: int

@app.post("/leaky", response_model=ChatResponse)
def leaky(body: ChatRequest):
    return {"answer": "ok", "tokens_used": 3, "internal_api_key": "sk-SECRET"}

Run it, and the secret never leaves the server:

200 {'answer': 'ok', 'tokens_used': 3}

That is a real, small security property, not a nice-to-have: a bug that accidentally attaches an internal key, a database row's password hash, or another user's field to a dict never reaches the client, because response_model only lets declared fields through the door. In Express, that same bug ships straight to the browser in the response body, because res.json(dbRow) sends whatever object you hand it -- there is nothing standing at the exit checking the shape.

And because the pydantic models ARE the source of truth, FastAPI can generate a live, browsable API reference from them for free -- visit /docs (interactive Swagger UI -- try a request from the browser itself) or fetch /openapi.json (the raw machine-readable schema other tools consume, including the client-generation tools some frontends use). No annotation comments, no separate YAML file, no drift -- both come from the exact function signatures and pydantic classes already sitting in your route file.

🌍 Real-world example: a hiring platform's /chat endpoint takes a candidate's message and a temperature setting. A pydantic model rejects a request missing the message with a precise 422 before any code runs, response_model guarantees the response never accidentally includes an internal field the frontend has no business seeing, and /docs lets a teammate try the endpoint from the browser without opening Postman.

💡 Path parameter = part of the URL itself (/items/{item_id}), always required, typed by the function signature.

💡 Query parameter = a plain function parameter that is not in the path and not the body (q: str | None = None); FastAPI infers it from where it is NOT declared.

💡 pydantic model = a Python class (subclass of BaseModel) that declares a shape once and gets validation, serialization, and docs generation from that single declaration.

💡 response_model = the declared shape of what a route is allowed to send back; anything the handler returns outside that shape is silently dropped before it reaches the client.

💡 uvicorn = the ASGI server process that actually runs a FastAPI app and listens on a port -- FastAPI is the framework, uvicorn is what executes it.

When to use it: pick FastAPI over hand-rolled Express validation whenever a route accepts a body with more than one or two fields, or whenever multiple teammates or a separate frontend team need an always-accurate reference for what an endpoint accepts and returns -- which is nearly every endpoint in an AI backend, since almost every route there takes a JSON body (a chat message, a document, settings) and needs its shape to be unambiguous to whoever calls it.

When NOT to use it / Trade-off: if you are writing one tiny internal script with a single fixed-shape call, hand-rolled validation costs you nothing extra and pulling in a full pydantic model is not buying you much. The trade-off is not "FastAPI is faster than Express" -- that is a performance claim neither this course nor you have measured, and you should never repeat it without your own benchmark. The real trade-off is where the validation work goes: in Express you write and maintain it by hand across the life of the endpoint; in FastAPI you write it once, as data (the pydantic class), and the framework maintains the checking and the docs for you from that data every time the shape changes.

Standard definition: FastAPI is a Python web framework, run by an ASGI server such as uvicorn, in which request and response shapes are declared once as pydantic (BaseModel) classes; FastAPI uses that single declaration to validate incoming path/query/body data (returning a structured 422 response with a loc-tagged error on failure), to filter outgoing responses to exactly the fields declared on a route's response_model, and to generate an interactive /docs UI and a machine-readable /openapi.json schema -- all from the same source of truth, with nothing written twice.

from fastapi import FastAPI
from fastapi.testclient import TestClient
from pydantic import BaseModel, Field

app = FastAPI()


# ---- path + query params (no pydantic needed for these) ----
@app.get("/items/{item_id}")
def read_item(item_id: int, q: str | None = None):
    return {"item_id": item_id, "q": q}


# ---- pydantic request model = the request body's shape + validation ----
class ChatRequest(BaseModel):
    message: str
    temperature: float = Field(default=0.7, ge=0.0, le=2.0)


class ChatResponse(BaseModel):
    answer: str
    tokens_used: int


@app.post("/chat", response_model=ChatResponse)
def chat(body: ChatRequest):
    return {"answer": f"echo: {body.message}", "tokens_used": len(body.message.split())}


# response_model as an output CONTRACT -- the handler leaks a fake secret,
# but only the fields declared on ChatResponse survive.
@app.post("/leaky", response_model=ChatResponse)
def leaky(body: ChatRequest):
    return {
        "answer": "ok",
        "tokens_used": 3,
        "internal_api_key": "sk-SECRET",
    }


client = TestClient(app)

print("== path + query params ==")
r = client.get("/items/42?q=hello")
print(r.status_code, r.json())

print()
print("== valid request ==")
r = client.post("/chat", json={"message": "hello world", "temperature": 0.5})
print(r.status_code, r.json())

print()
print("== missing field ==")
r = client.post("/chat", json={"temperature": 0.5})
print(r.status_code, r.json())

print()
print("== out-of-range temperature (Field(ge=0.0, le=2.0), sent 5) ==")
r = client.post("/chat", json={"message": "hi", "temperature": 5})
print(r.status_code, r.json()["detail"][0])

print()
print('== handler returned {"answer","tokens_used","internal_api_key": "sk-SECRET"} ==')
r = client.post("/leaky", json={"message": "hi", "temperature": 0.5})
print(r.status_code, r.json(), "        <- the secret was DROPPED by response_model")

print()
print("== auto docs ==")
schema = client.get("/openapi.json").json()
print("paths:", list(schema["paths"].keys()))
print("schemas:", list(schema["components"]["schemas"].keys()))
r = client.get("/docs")
print("GET /docs ->", r.status_code)

What we're building: a multi-user AI chat backend in FastAPI -- the kind of project a resume calls "built an AI SaaS backend" and an interview actually probes. Every user gets their own history, their own quota, and their own cache slot; the model call itself streams back over Server-Sent Events; and a slow document upload runs as a background job you poll instead of a request that hangs for a second. This project is FREE, but everything that explains WHY each piece works the way it does (topics 3-13) is locked -- so every step below is written to stand on its own, re-deriving each check inline rather than pointing back at a topic you have not seen.

🌍 Real-world example: think of a real product like a hosted "AI resume reviewer" API two different companies both pay to use. Company A's requests must never see Company B's cached answers, Company A running out of its monthly quota must not block Company B, and a slow PDF upload from either company should not tie up the same server thread that is answering everyone else's chat requests right now. That is this project, end to end.

💡 multi-tenant = many independent users/customers sharing one running application and one set of servers, where one tenant's data and limits must never leak into another's. 💡 request path = the ordered list of steps one HTTP request passes through before a reply goes back: here it is auth -> quota -> cache -> (fake) model -> log -> response.

Step 1 -- The app skeleton and the pydantic request/response models

Start with the shapes, before any logic. ChatRequest is what a caller must send; ChatResponse is what your own handler is allowed to send back:

from fastapi import FastAPI
from pydantic import BaseModel, Field

app = FastAPI()

class ChatRequest(BaseModel):
    message: str
    system_prompt: str = "You are a helpful assistant."
    model: str = "small-chat-model"
    temperature: float = Field(default=0.0, ge=0.0, le=2.0)

class ChatResponse(BaseModel):
    answer: str
    tokens_used: int
    cache_status: str

Why this is Step 1, before a single route: pydantic validates every incoming body against ChatRequest for free -- a request missing message, or sending temperature=5 when the field says le=2.0, never reaches your function body at all; FastAPI returns a 422 with a loc array naming exactly which field was wrong, not a generic 400. And declaring the route with response_model=ChatResponse (used below) is an OUTPUT contract: if your handler's dict ever grows an extra key by accident -- an internal id, a debug field -- pydantic drops it before it reaches the client. Validation and the response shape are not something you write if checks for; they are something you declare once, as a type.

Step 2 -- JWT auth as a dependency; every route gets a real user_id

import jwt
from fastapi import Depends, Header, HTTPException

SECRET_KEY = "demo-secret-do-not-use-in-prod"
ALGORITHM = "HS256"

def current_user(authorization: str | None = Header(default=None)) -> str:
    if authorization is None or not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Not authenticated")
    token = authorization.removeprefix("Bearer ")
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token expired")
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="Invalid token")
    return payload["sub"]

Why a dependency, not a manual check inside every handler: once a route's function signature reads user_id: str = Depends(current_user), that handler CANNOT be written without a user_id -- there is no path through the code that skips it, unlike a hand-written if not token: ... at the top of ten different functions, one of which someone eventually forgets. And the two separate except clauses matter: jwt.ExpiredSignatureError and jwt.InvalidTokenError are genuinely different situations -- a bare except: would collapse "this token used to be valid and timed out" and "this token is forged or corrupt" into the same message, and would also swallow KeyboardInterrupt.

One fact worth stating plainly: a JWT is signed, not encrypted. Anyone holding the token can decode its payload without the secret key -- the secret only proves the payload was not tampered with. Never put anything in a JWT payload you would not print on a postcard.

When to use it: any endpoint that needs to know WHOSE data it is touching -- which, once there is more than one user, is every endpoint in this project. When NOT to use it: a genuinely public, unauthenticated endpoint (a health check, a public pricing page) should not carry Depends(current_user) at all -- forcing auth on a route that has no user-specific data to protect just breaks it for anonymous callers with nothing gained.

Step 3 -- Per-user chat history (keyed by user_id)

chat_history: dict[str, list[dict]] = {}

# inside the /chat handler, after computing `answer` for this user_id:
history = chat_history.setdefault(user_id, [])
history.append({"role": "user", "content": req.message})
history.append({"role": "assistant", "content": answer})

Why keyed by user_id and nothing else: the dictionary's key is the ENTIRE isolation mechanism here. chat_history["user_alice"] and chat_history["user_bob"] are two completely separate lists purely because they are two different dict keys -- there is no code anywhere that explicitly says "don't let alice see bob's messages"; the isolation is a side effect of the data structure's shape. That is also exactly why forgetting to include user_id ANYWHERE in this project (the cache in Step 5 is the one people forget) breaks the same guarantee. user_id comes from Depends(current_user) in Step 2 -- it is never read from the request body, because a client could put anything it wants in a body field, but it cannot forge a JWT it does not hold the secret for.

Step 4 -- A quota check BEFORE the model call (after it, you have already paid)

FREE_PLAN_QUOTA = 3
usage_counts: dict[str, int] = {}

def check_quota(user_id: str):
    used = usage_counts.get(user_id, 0)
    if used >= FREE_PLAN_QUOTA:
        raise HTTPException(
            status_code=429,
            detail=f"quota exceeded: {used}/{FREE_PLAN_QUOTA} requests used on the free plan. Upgrade to continue.",
        )

def record_usage(user_id: str):
    usage_counts[user_id] = usage_counts.get(user_id, 0) + 1

# inside the handler, in THIS order:
check_quota(user_id)      # 1. check the budget
# ... cache lookup, then the model call only runs on a MISS ...
record_usage(user_id)     # 2. only count it once an answer was actually served

Why the ORDER is the entire lesson: check_quota runs before the cache lookup and before the model would ever be called. If you checked the quota AFTER calling the model, you would already have spent the tokens for the request you are about to refuse -- the check would be a receipt, not a limit. A 429 status code (not 403, not a silent empty response) tells the CALLER's code "this is a rate/plan limit, not a permissions problem or a crash," and the detail message gives a human an upgrade path instead of a mystery failure.

When to use it: any endpoint that costs you money or a shared resource per call -- which is every route in this project that reaches the model. When NOT to use it / trade-off: this is a product/billing concern, separate from rate limiting as a SECURITY control (throttling abusive traffic regardless of plan) -- that is a different chapter's job; do not conflate the two or you end up enforcing a business rule with a security tool or vice versa.

Step 5 -- A response cache whose key includes the user id (leave it out and you leak across tenants)

import json, hashlib

response_cache: dict[str, dict] = {}

def cache_key(user_id: str, model: str, system_prompt: str, message: str) -> str:
    payload = {
        "user_id": user_id,   # <-- leave this out and tenant B reads tenant A's answer
        "model": model,
        "system_prompt": system_prompt,
        "message": message,
    }
    blob = json.dumps(payload, sort_keys=True)   # sort_keys: same dict, same key, regardless of field order
    return hashlib.sha256(blob.encode()).hexdigest()

Why FOUR fields and not just the message: message alone is not enough of a key -- two requests can carry the identical message and legitimately need two different cached answers. system_prompt must be in the key because a different persona/instruction set changes what the correct answer even is; leave it out and one persona's cached reply gets served under another's instructions. model must be in the key for the same reason -- a cheap and an expensive model can answer the same prompt differently. And user_id is the one people forget, because a cache "obviously" feels like infrastructure, not user data -- but this project actually ran two users sending the identical message, and it is user_id inside the hash that is the ONLY reason bob's call was a MISS instead of silently returning alice's cached answer (verified in the output below). sort_keys=True matters too: json.dumps without it can produce a different string (and therefore a different hash) for the exact same dict depending on field insertion order.

When to use it: repeated, non-personalised, not-time-sensitive requests, where paying for the same answer twice is pure waste -- a cache HIT here measures at a few hundred microseconds against a model call that (with a real provider) takes hundreds of milliseconds or more. When NOT to use it: a genuinely personalised answer, anything time-sensitive (today's date, "what changed since I last asked"), or any request where temperature is set above 0 SPECIFICALLY because the caller wants variety -- caching there does not save money, it silently breaks the feature.

Step 6 -- The streaming chat endpoint (SSE)

from fastapi.responses import StreamingResponse

@app.post("/chat/stream")
def chat_stream(req: ChatRequest, user_id: str = Depends(current_user)):
    check_quota(user_id)
    record_usage(user_id)

    def event_generator():
        try:
            for word in fake_llm_stream(req.message):
                yield f"data: {json.dumps({'delta': word})}\n\n"
            yield "data: [DONE]\n\n"
        except Exception as exc:
            yield f"data: {json.dumps({'error': str(exc)})}\n\n"

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )

Why stream at all -- and the one thing to never claim about it: streaming buys you a faster TIME-TO-FIRST-TOKEN -- the caller sees the first word almost immediately instead of waiting for the whole answer to finish generating. It does NOT make the answer arrive faster overall -- total generation time is usually the same or slightly worse than a single blocking call; do not claim otherwise in an interview.

The non-obvious half of "just add streaming": the HTTP status code (200) and the headers are sent to the client the moment the FIRST chunk goes out -- verified above, the response starts streaming data: frames before a single word of the answer existed. If something fails while generating word 5 of 10, you cannot go back and turn that 200 into a 500 -- the status line is already gone. The except block inside event_generator is not decoration; it is the ONLY remaining place an error can be reported once the stream has started, as one more data: frame the frontend has to check for.

When to use it: a one-way flow of tokens from server to a single caller, which is what a chat reply is. When NOT to use it / trade-off: SSE is one-way (server -> client) and rides on plain HTTP; if you need the CLIENT to also push messages back on the same connection (a live collaborative editor, bidirectional presence), that is a WebSocket's job, not SSE's -- do not reach for a WebSocket for a one-way token stream, it adds reconnect and scaling complexity you do not need here.

Step 7 -- Document upload as a background job + a status endpoint to poll

from fastapi import BackgroundTasks

jobs: dict[str, dict] = {}

def process_document(job_id: str, filename: str):
    jobs[job_id]["status"] = "running"
    time.sleep(1.0)   # stand-in for parsing/chunking/embedding
    jobs[job_id]["status"] = "done"
    jobs[job_id]["chunks"] = 12

@app.post("/documents")
def upload_document(filename: str, background_tasks: BackgroundTasks,
                     user_id: str = Depends(current_user)):
    job_id = f"job_{len(jobs) + 1}"
    jobs[job_id] = {"job_id": job_id, "status": "queued", "chunks": None, "user_id": user_id}
    background_tasks.add_task(process_document, job_id, filename)
    return jobs[job_id]

@app.get("/documents/{job_id}")
def document_status(job_id: str, user_id: str = Depends(current_user)):
    job = jobs.get(job_id)
    if job is None:
        raise HTTPException(status_code=404, detail="job not found")
    return job

Why the response returns before the work is done: background_tasks.add_task(...) schedules process_document to run AFTER the HTTP response has already been sent -- verified below: POST /documents came back in 9 ms while the actual 1-second job was still running underneath it. Holding the request open for a full second (or longer, for a real document) is exactly what causes proxy timeouts and loses the work if the caller's connection drops -- accept-then-poll avoids both. GET /documents/{job_id} is a plain read of the same in-memory dict, polled by the caller until status becomes "done".

The three limits this does NOT solve (say these unprompted): jobs is a plain dict in ONE process's RAM -- a second worker process serving a different request would have no idea this job exists; there is no retry, so if process_document raises, the job silently stays "running" forever with no error surfaced anywhere; and there is no concurrency cap, so 100 simultaneous uploads start 100 background tasks with nothing throttling them. Those three gaps are exactly what a real task queue (Celery, RQ, a managed cloud queue) exists to close -- this project stops deliberately at the point where you would reach for one, and describes it rather than faking it, because none of those packages are installed here.

When to use it: anything slower than a request should reasonably wait for -- document ingestion, batch scoring, report generation. When NOT to use it / trade-off: anything the CALLER needs the answer to immediately (the /chat endpoint itself) has no business being a background job -- accept-then-poll trades an instant answer for a delayed one, which is wrong for the primary chat path.

Step 8 -- Structured logging of tokens, latency and cache status, and a final tally

request_log: list[dict] = []

def log_request(user_id, endpoint, tokens, latency_ms, cache_status):
    entry = {"user_id": user_id, "endpoint": endpoint, "tokens": tokens,
              "latency_ms": round(latency_ms, 1), "cache_status": cache_status}
    request_log.append(entry)
    print("LOG", json.dumps(entry))

@app.get("/admin/tally")
def tally():
    total_tokens = sum(e["tokens"] for e in request_log)
    hits = sum(1 for e in request_log if e["cache_status"] == "HIT")
    total = len(request_log)
    return {
        "total_requests": total,
        "total_tokens": total_tokens,
        "cache_hit_rate": round(hits / total, 2) if total else 0.0,
        "requests_per_user": {u: sum(1 for e in request_log if e["user_id"] == u)
                                for u in {e["user_id"] for e in request_log}},
    }

Why log a structured dict and not a sentence: print(f"user {user_id} used {tokens} tokens") is readable but not QUERYABLE -- you cannot ask "what was my cache hit rate yesterday" of a pile of sentences. A structured entry (json.dumps(entry)) is what a real logging/observability stack (this project just prints it; a real one ships it to a log aggregator) can filter, group and aggregate on. The /admin/tally endpoint above IS that aggregation, done by hand on the in-memory list: verified in this project's own run, four logged requests produced a real cache_hit_rate of 0.25 (one HIT out of four) and a real per-user breakdown -- this is the same shape as a resume line that says "measured a 25% cache hit rate" instead of "added caching."

Where a real provider call replaces fake_llm (illustrative)

# ---- ILLUSTRATIVE (requires an API key; not executed) ----
# Everywhere above, fake_llm(req.message) stands in for this. Swapping it in changes
# nothing about auth, quota, the cache key, streaming, or the background job -- it is
# one function body, called from the exact same places.
def real_llm(message: str, system_prompt: str) -> str:
    response = provider_client.chat.completions.create(
        model="<current-small-chat-model>",  # model IDs change -- check the provider's current list
        messages=[{"role": "system", "content": system_prompt},
                  {"role": "user", "content": message}],
    )
    return response.choices[0].message.content

What to say about this in an interview (the honest version)

What this project genuinely proves: you can build every station on the request path a real AI backend needs -- auth (a JWT dependency every route depends on), identity-scoped state (per-user history AND a per-user quota AND a cache key that includes the user id, all three verified not to leak across users), cost control before the call (quota checked before the model runs, not after), a streaming endpoint that hands back Server-Sent Events instead of one blocking response, work moved off the request path (a background job with a pollable status endpoint, measured at 9 ms instead of blocking for a second), and structured logs feeding a tally (tokens, latency, cache status, per-user counts). Every number in this project's output came from a real running server, not a description of one.

What it does NOT do yet, and you should say so before an interviewer finds it:

• Single process, in-memory stores. chat_history, usage_counts, response_cache and jobs are all plain Python dicts living in this one process's RAM. Restart the process and every user's history, quota counter, cache and job status are gone. A second worker process (which you would want for real traffic) would not see any of them either -- each worker has its own copy.
• No real queue. BackgroundTasks runs in the same web process. There is no retry if process_document raises, no concurrency limit if 100 uploads arrive at once, and no cross-worker visibility. That is exactly what a real task queue (Celery, RQ, a managed cloud queue) exists to fix -- this project deliberately stops at the point where you would reach for one.
• The model call is illustrative, not run. fake_llm / fake_llm_stream are deterministic stand-ins so the entire request path -- auth, quota, cache, streaming, background jobs, logging -- runs for real with zero API keys. Swapping in a real provider call changes nothing about the wiring above it; it only replaces one function body.
• No database. A real product needs the history, quota counters and cache to survive a restart -- that means a relational database or Redis, not a dict. This project teaches the SHAPE each of those needs to have (what the key is, when it is checked, what it must contain); persisting it is the next step, not this one.

Naming these limits unprompted is what separates "I built a chatbot" from "I built the parts of a multi-user backend and I know exactly where it stops being production-ready."

Standard definition: a multi-user AI chat backend authenticates every request with a dependency-injected identity, scopes history/quota/cache to that identity (the cache key is the one people forget), checks budget before spending it, streams the model's output over Server-Sent Events, and moves slow work (like document ingestion) off the request path behind a job id you can poll -- all of which this project runs for real, in one process, with in-memory state, which is the honest starting point before a database and a real queue.

import time
import json
import hashlib

import jwt
from fastapi import FastAPI, Depends, HTTPException, Header, BackgroundTasks
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field

app = FastAPI()
SECRET_KEY = "demo-secret-do-not-use-in-prod"
ALGORITHM = "HS256"


# ---- Step 1: request/response models ----
class ChatRequest(BaseModel):
    message: str
    system_prompt: str = "You are a helpful assistant."
    model: str = "small-chat-model"
    temperature: float = Field(default=0.0, ge=0.0, le=2.0)


class ChatResponse(BaseModel):
    answer: str
    tokens_used: int
    cache_status: str  # "HIT" or "MISS"


# ---- Step 2: JWT auth as a dependency ----
def current_user(authorization: str | None = Header(default=None)) -> str:
    if authorization is None or not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Not authenticated")
    token = authorization.removeprefix("Bearer ")
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token expired")
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="Invalid token")
    return payload["sub"]


# ---- Step 3: per-user chat history ----
chat_history: dict[str, list[dict]] = {}

# ---- Step 4: quota store ----
FREE_PLAN_QUOTA = 3
usage_counts: dict[str, int] = {}


def check_quota(user_id: str):
    used = usage_counts.get(user_id, 0)
    if used >= FREE_PLAN_QUOTA:
        raise HTTPException(
            status_code=429,
            detail=f"quota exceeded: {used}/{FREE_PLAN_QUOTA} requests used on the free plan. Upgrade to continue.",
        )


def record_usage(user_id: str):
    usage_counts[user_id] = usage_counts.get(user_id, 0) + 1


# ---- Step 5: response cache -- the key includes the user id ----
response_cache: dict[str, dict] = {}


def cache_key(user_id: str, model: str, system_prompt: str, message: str) -> str:
    payload = {
        "user_id": user_id,   # <-- leave this out and tenant B reads tenant A's answer
        "model": model,
        "system_prompt": system_prompt,
        "message": message,
    }
    blob = json.dumps(payload, sort_keys=True)
    return hashlib.sha256(blob.encode()).hexdigest()


# fake "LLM" -- deterministic, no API key needed, so the whole flow really runs
def fake_llm(message: str) -> str:
    return f"echo: {message}"


def fake_llm_stream(message: str):
    for word in f"echo: {message}".split(" "):
        yield word


def count_tokens(text: str) -> int:
    return len(text.split())


request_log: list[dict] = []


def log_request(user_id, endpoint, tokens, latency_ms, cache_status):  # Step 8
    entry = {"user_id": user_id, "endpoint": endpoint, "tokens": tokens,
              "latency_ms": round(latency_ms, 1), "cache_status": cache_status}
    request_log.append(entry)
    print("LOG", json.dumps(entry))


@app.post("/chat", response_model=ChatResponse)
def chat(req: ChatRequest, user_id: str = Depends(current_user)):
    start = time.perf_counter()
    check_quota(user_id)                                    # Step 4 -- BEFORE the model call

    key = cache_key(user_id, req.model, req.system_prompt, req.message)   # Step 5
    cached = response_cache.get(key)
    if cached:
        cache_status, answer, tokens = "HIT", cached["answer"], cached["tokens"]
    else:
        cache_status = "MISS"
        answer = fake_llm(req.message)                       # <-- real provider call goes here
        tokens = count_tokens(req.message) + count_tokens(answer)
        response_cache[key] = {"answer": answer, "tokens": tokens}

    record_usage(user_id)                                    # pay only once we actually served an answer

    history = chat_history.setdefault(user_id, [])           # Step 3
    history.append({"role": "user", "content": req.message})
    history.append({"role": "assistant", "content": answer})

    latency_ms = (time.perf_counter() - start) * 1000
    log_request(user_id, "/chat", tokens, latency_ms, cache_status)
    return ChatResponse(answer=answer, tokens_used=tokens, cache_status=cache_status)


# ---- Step 6: streaming chat endpoint (SSE) ----
@app.post("/chat/stream")
def chat_stream(req: ChatRequest, user_id: str = Depends(current_user)):
    check_quota(user_id)
    record_usage(user_id)

    def event_generator():
        try:
            for word in fake_llm_stream(req.message):
                yield f"data: {json.dumps({'delta': word})}\n\n"
            yield "data: [DONE]\n\n"
        except Exception as exc:
            # the ONLY place left to report a failure once the 200 is already on the wire
            yield f"data: {json.dumps({'error': str(exc)})}\n\n"

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )


# ---- Step 7: document upload as a background job + status endpoint ----
jobs: dict[str, dict] = {}


def process_document(job_id: str, filename: str):
    jobs[job_id]["status"] = "running"
    time.sleep(1.0)                                          # stand-in for parsing/chunking/embedding
    jobs[job_id]["status"] = "done"
    jobs[job_id]["chunks"] = 12


@app.post("/documents")
def upload_document(filename: str, background_tasks: BackgroundTasks, user_id: str = Depends(current_user)):
    job_id = f"job_{len(jobs) + 1}"
    jobs[job_id] = {"job_id": job_id, "status": "queued", "chunks": None, "user_id": user_id}
    background_tasks.add_task(process_document, job_id, filename)
    return jobs[job_id]


@app.get("/documents/{job_id}")
def document_status(job_id: str, user_id: str = Depends(current_user)):
    job = jobs.get(job_id)
    if job is None:
        raise HTTPException(status_code=404, detail="job not found")
    return job


# ---- Step 8: the final tally ----
@app.get("/admin/tally")
def tally():
    total_tokens = sum(e["tokens"] for e in request_log)
    hits = sum(1 for e in request_log if e["cache_status"] == "HIT")
    total = len(request_log)
    return {
        "total_requests": total,
        "total_tokens": total_tokens,
        "cache_hit_rate": round(hits / total, 2) if total else 0.0,
        "requests_per_user": {u: sum(1 for e in request_log if e["user_id"] == u)
                                for u in {e["user_id"] for e in request_log}},
    }