Full-Stack AI interview questions & answers

Ek ghar ki kitchen aur ek restaurant ki kitchen ka farak socho. Ghar pe, ek insaan jo bhi room me khada hai uske liye khana banata hai, jo bhi fridge me mila usi se, aur agar stove kharab ho jaaye toh bas aaj ka dinner late ho jaayega. Restaurant ki kitchen ek saath chalis tables ke liye khana banaati hai, use exactly pata hona chahiye ki konsi dish kis table ki hai, har ingredient ka cost track karti hai, aur us raat ke liye plan hota hai jab service ke beech me gas line fail ho jaaye. Same stove, same recipes -- bilkul alag kaam.

Isse pehle ka har chapter ghar ki kitchen tha. RAG (Ch4), tool calling (Ch5), LangChain agents (Ch6), memory (Ch7), fine-tuning (Ch8), guardrails (Ch9) -- ye sab ek user, ek machine, ek time pe ek request ki tarah chala, ek API key chupchap ek variable me baithi hui aur aakhir me ek print(). Ye ek script hai, aur script ko in char problems ko ignore karne ki chhoot hoti hai kyunki room me hamesha sirf ek hi insaan hota hai jo unse affect ho sakta hai.

Jaise hi ek dusra user aata hai, ye char problems optional nahi rehti:

• Concurrency -- kai requests paas-paas aati hain, kabhi-kabhi bilkul same instant pe. Ek script ek ke baad ek cheez process karti hai; ek product ko decide karna padta hai ki jab request one abhi chal hi rahi hai tab request two ka kya hoga.
• Identity -- "ye data kiska hai?" Script ka jawab hamesha "mera" tha aur ye sawaal kabhi poochna hi nahi pada. Ek product ko har single request ke liye pata hona chahiye ki konsa user pooch raha hai -- kyunki unki chat history, unke documents, unka usage, aur (is chapter ke baad ke topics jaise dikhaayenge) unke cached answers bhi isi pe depend karte hain ki ye sahi ho.
• Cost -- koi na koi per-token pay kar raha hai, har call pe, har user se, poore din. Ek script ek experiment hai; ek product ek chalta hua bill hai jo aapke sote hue bhi tab rakhta hai.
• Failure -- model provider internet ke paar ek remote service hai. Ye timeout hoga, error return karega, ya response ke beech me hi fail ho jaayega, aur ye politely aapke convenient moment ka wait nahi karega.

In char me se koi bhi prompt ke baare me nahi hai. Aapke paas duniya ka best prompt ho sakta hai aur phir bhi aap aisa backend ship kar sakte hain jo har user ke liye freeze ho jaaye jaise hi ek slow request aati hai, ya ek customer ki chat history dusre customer ki screen pe leak ho jaaye. Ye chapter model call ke aas-paas ki plumbing ke baare me hai -- wo hissa jo ek notebook ko kabhi zaroorat nahi padi aur ek product uske bina ship nahi ho sakta.

Us plumbing ko sahi rakhne ke liye, is poore chapter ke liye ek diagram apne dimaag me rakho -- the request path:

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

Is chapter ka har baad ka topic is path ka ek station hai: fastapi-basics aur async-vs-sync-endpoints batate hain ki server request ko accept kaise karta hai; auth-multitenancy auth box hai; usage-metering-quotas quota box hai; caching-llm-responses cache box hai; streaming-sse-backend / vercel-ai-sdk / streaming-ui-frontend stream back box hain; error-handling-timeouts batata hai ki jab provider box galat behave kare toh kya hota hai; background-jobs-queues us kaam ke liye hai jo is path pe kabhi baithna hi nahi chahiye; aur ai-backend-architecture poori tasveer banaata hai sab stations ke saath. Beech ka [ RAG / tools / memory ] box koi naya kaam nahi hai -- ye exactly Ch4, Ch5, Ch6 aur Ch7 hi hai, is path ke andar se ek function ki tarah call kiya hua. Ye chapter us box ke andar jo hota hai wo dobara nahi sikhaata; ye uske aas-paas sab kuch sikhata hai.

Neeche wala snippet us jump ka sabse chhota possible version concrete banaata hai: wahi fake model call, pehle ek plain script ki tarah chalaya gaya (bilkul har pichle chapter jaisa), phir wahi call ek FastAPI endpoint ke peeche TestClient use karke dobara chalaya gaya. Dhyan se dekho ki kya badla aur kya nahi -- honest jawab hai "jitna aap sochte hain usse kam," aur yehi gap is chapter ka asli subject hai.

🌍 Real-world example: ek college project jo ek PDF ko ek user ke liye summarize karta hai, Jupyter cell se chalaaya gaya, ek script hai. Jaise hi aap use deploy karte hain taaki classmates apni-apni PDF upload karke saath-saath apna-apna summary wapas paayein, aapko concurrency (ek badi PDF baaki sabki request ko freeze na kar de), identity (ye kiski PDF hai aur kiska summary), cost (kisi ka account per-upload bill ho raha hai), aur failure handling (agar model call timeout ho jaaye toh classmate ko kya dikhega) sab dena padega -- jab tak sirf aap tha, ek cell me, ek baar chala raha tha, ye kuch bhi problem nahi tha.

💡 Script = wo code jo ek user, ek time pe ek request serve karta hai, "meri call" aur "kisi aur ki call" ke beech koi separation nahi hoti kyunki koi aur hota hi nahi.

💡 Product = wahi core logic, ab kai users ek saath reach kar sakte hain, har ek ko apni identity chahiye, apna cost track chahiye, aur ek defined jawab chahiye ki jab kuch fail ho toh unhe kya dikhega.

💡 The request path = wo fixed sequence jisse har request guzarti hai: browser -> auth -> quota -> cache -> RAG/tools/memory -> provider -> stream back -> log. Is chapter ka har topic isi path ka ek station hai.

Kab use karein: kisi project ko "bas ek script" tab tak treat karo jab tak use ek time pe exactly ek hi insaan call karega -- ek personal tool, ek one-off data job, ek demo jo aap khud chalata hai. Jaise hi ek doosra, independent caller exist karta hai (ek teammate, ek classmate, ek paying user), aap ek product bana raha hai chahe tune waise socha ho ya nahi, aur upar wali char problems already live hain.

Kab NAHI use karein (over-engineer mat karo): din ek pe hi auth, cache layer, aur job queue ki zaroorat nahi hai. Honest early architecture ek process hai jo already "kaun pooch raha hai" ko "unhone kya poocha" se alag rakhta hai, taaki baad me real auth ya real cache add karna ek chhota change ho, poora rewrite nahi. Ek doosra real user aane se pehle hi teen services bana dena bhi apni ek mistake hai -- is chapter ke baad ka ai-backend-architecture topic exactly isi trade-off pe wapas aata hai.

Standard definition (interview me bolo): 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")

Ek restaurant order counter socho. Express me, ye counter aap khud banate the: aap khud wo sign likhte the jisme bataya jaata tha ki ek valid order kaisa dikhta hai, aap khud check karte the ki customer ne order slip ke har field me kuch bhara hai ya nahi, aur agar customer ne quantity box me number ki jagah "five" likh diya toh aapke apne code ko ye notice karke reject karna padta tha. FastAPI ek counter clerk hire karta hai jise order form pehle se yaad hai: aap clerk ko bas ye bata dete ho ki ek valid order slip kaisa dikhna chahiye, aur clerk har order ko usse check kar leta hai isse pehle ki wo aap tak pahunche -- notice-and-reject wala apna code likhne ki zaroorat hi nahi.

FastAPI me path/query/body handling directly usi cheez pe map hoti hai jo aap Express se pehle se jaante ho -- same teen inputs, alag 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 (hamesha ek string, aap khud convert karte ho)	q: str | None = None ek function parameter ke roop me -- typed aur default value ki wajah se by default optional
req.body, phir express-validator ya haath se likha if (!req.body.message)	ek pydantic request model: class ChatRequest(BaseModel): message: str
npm install swagger-jsdoc swagger-ui-express + har route ke upar YAML comments likhna	/docs aur /openapi.json automatically exist karte hain, usi code se generate hoke jo aap already likh chuke ho
node server.js / nodemon	uvicorn main:app --reload -- wo ASGI server jo actually aapki FastAPI app ko chalata hai; FastAPI framework hai, uvicorn use execute karta hai, waise hi jaise Express framework hai aur Node use chalata hai

Jo genuinely alag hai, sirf naam badla hua nahi: Express me, validation aur docs do alag libraries hain jo aap bolt-on karte ho -- ek ke liye express-validator, doosre ke liye swagger-jsdoc -- aur dono aapke actual route code se silently out of sync ho sakte hain, kyunki kuch bhi unhe agree karne pe force nahi karta. FastAPI me, pydantic model HI validation hai, aur wahi model HI documentation hai. Aap shape ek baar declare karte ho; FastAPI dono ko usi ek source of truth se generate kar deta hai. Ek alag schema file yaad rakhne ki zaroorat hi nahi hoti.

Yahaan request model aapke liye checking kar raha hai:

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

message ke bina ek request bhejo, aur FastAPI aapke function ko call bhi nahi karta -- wo khud request ko reject kar deta hai:

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

ge=0.0, le=2.0 declare kiye gaye field ke against temperature: 5 bhejo, aur wahi cheez hoti hai:

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

Teen cheezein notice karne wali hain, aur ye bilkul wahi hain jo ek Express developer ko shift karte waqt confuse karti hain. Pehla, ye 422 Unprocessable Entity hai, 400 Bad Request nahi -- 400 ka matlab hai ki request khud hi malformed hai (broken JSON); 422 ka matlab hai ki JSON parse toh sahi hui, lekin declared shape ko satisfy nahi kiya. Agar aapka frontend sirf if (status === 400) check karta hai toh wo har validation failure ko silently miss kar dega. Doosra, loc array bilkul batata hai ki konsa field fail hua -- ['body', 'temperature'] -- jo bilkul wahi hai jo ek frontend ko sahi input box ko red highlight karne ke liye chahiye, aur isse produce karne ke liye aapne ek line bhi code nahi likha. Teesra, ye is model ko use karne wale har endpoint ke liye automatically hota hai -- upar wale handler me kahin bhi if not request.body.message: return 400 nahi hai. Shape ek baar declare karo; checking hamesha ke liye free ho jaati hai.

Ab doosri direction: response_model documentation nahi hai, ye OUT jaane waale rasta pe ek filter hai. Declare karo ki ek response me kya hona chahiye, aur FastAPI handler jo bhi return kare usme se jo bhi us shape me nahi hai use hata deta hai -- chaahe handler ne khud galti se use daala ho:

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"}

Ise chalao, aur secret server se kabhi bahar nahi jaata:

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

Ye ek real, chhota security property hai, sirf ek nice-to-have nahi: ek bug jo galti se ek internal key, ek database row ka password hash, ya kisi doosre user ka field attach kar de, wo client tak kabhi nahi pahunchega, kyunki response_model sirf declared fields ko hi darwaaze se guzarne deta hai. Express me, wahi bug seedhe browser tak response body me pahunch jaata, kyunki res.json(dbRow) jo bhi object aap use pakadate ho wahi bhej deta hai -- exit pe koi shape check karta hua khada nahi hota.

Aur kyunki pydantic models HI source of truth hain, FastAPI unse ek live, browse karne laayak API reference free me generate kar sakta hai -- /docs pe jao (interactive Swagger UI -- browser se hi ek request try karo) ya /openapi.json fetch karo (raw machine-readable schema jise dusre tools consume karte hain, un client-generation tools sameet jo kuch frontends use karte hain). Koi annotation comments nahi, koi alag YAML file nahi, koi drift nahi -- dono usi route file me baithe function signatures aur pydantic classes se aate hain.

🌍 Real-world example: ek hiring platform ke /chat endpoint ek candidate ka message aur ek temperature setting leta hai. Ek pydantic model message ke bina aayi request ko kisi bhi code chalne se pehle ek precise 422 ke saath reject kar deta hai, response_model guarantee karta hai ki response me kabhi galti se koi internal field na aa jaaye jise frontend ko dekhne ka koi kaam nahi, aur /docs ek teammate ko Postman khole bina browser se endpoint try karne deta hai.

💡 Path parameter = URL ka hi ek hissa (/items/{item_id}), hamesha required, function signature se typed.

💡 Query parameter = ek plain function parameter jo path me nahi hai aur body me bhi nahi (q: str | None = None); FastAPI ise wahaan se infer karta hai jahan wo declare NAHI hua.

💡 pydantic model = ek Python class (BaseModel ki subclass) jo ek shape ek baar declare karti hai aur usi ek declaration se validation, serialization aur docs generation paati hai.

💡 response_model = wo declared shape jo ek route ko wapas bhejne ki ijaazat hai; handler jo bhi us shape se bahar return kare wo client tak pahunchne se pehle silently drop ho jaata hai.

💡 uvicorn = wo ASGI server process jo actually ek FastAPI app ko chalata hai aur ek port pe listen karta hai -- FastAPI framework hai, uvicorn use execute karta hai.

Kab use karein: hand-rolled Express validation ki jagah FastAPI tab choose karo jab ek route body me ek ya do se zyada fields accept kare, ya jab multiple teammates ya ek alag frontend team ko ek hamesha-accurate reference chahiye ho ki ek endpoint kya accept karta hai aur kya return karta hai -- jo ek AI backend me lagbhag har endpoint ka case hai, kyunki wahaan almost har route ek JSON body leta hai (ek chat message, ek document, settings) aur uski shape har caller ke liye unambiguous honi chahiye.

Kab NAHI use karein / Trade-off: agar aap ek chhota internal script likh rahe ho jisme ek fixed-shape call hai, toh hand-rolled validation kuch extra cost nahi karta aur ek poora pydantic model laana zyada faayda nahi deta. Trade-off ye nahi hai ki "FastAPI Express se fast hai" -- ye ek performance claim hai jo na is course ne na aapne measure ki hai, aur ise apni khud ki benchmark ke bina kabhi repeat mat karna. Asli trade-off ye hai ki validation ka kaam kahan jaata hai: Express me aap use endpoint ki poori life me haath se likhte aur maintain karte ho; FastAPI me aap ise ek baar data ki tarah likhte ho (pydantic class), aur jab bhi shape badalti hai, framework us data se checking aur docs dono ko aapke liye maintain karta hai.

Standard definition (interview me bolo): 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)

Kya bana rahe hain: FastAPI mein ek multi-user AI chat backend -- wo project jise resume pe "built an AI SaaS backend" likha jaata hai aur interview mein actually poocha jaata hai. Har user ki apni history hai, apna quota hai, aur apna cache slot hai; khud model call Server-Sent Events pe stream hoti hai; aur ek slow document upload ek background job ki tarah chalta hai jise aap poll karte hain, na ki ek request jo ek second ke liye hang ho jaaye. Ye project FREE hai, par wo sab kuch jo har cheez ke WHY explain karta hai (topics 3-13) LOCKED hai -- to neeche har step khud se complete likha hai, har check ko inline se-derive karta hai, kisi aisi topic ki taraf point kiye bina jo aapne dekhi hi nahi.

🌍 Real-world example: ek real product socho jaise ek hosted "AI resume reviewer" API jise do alag companies use karne ke liye pay karti hain. Company A ki requests ko kabhi Company B ke cached answers nahi dikhne chahiye, Company A ka monthly quota khatam hona Company B ko block nahi karna chahiye, aur kisi bhi company se ek slow PDF upload ko usi server thread ko busy nahi rakhna chahiye jo abhi sabke chat requests ka jawab de rahi hai. Yehi ye project hai, end to end.

💡 multi-tenant = kayi independent users/customers ek hi chal rahi application aur ek hi set of servers share karte hain, jahan ek tenant ka data aur limits kabhi doosre mein leak nahi hone chahiye. 💡 request path = ek HTTP request ke steps ka ordered list jo wo reply wapas jaane se pehle cross karti hai: yahan ye hai auth -> quota -> cache -> (fake) model -> log -> response.

Step 1 -- App skeleton aur pydantic request/response models

Shapes se shuru karo, kisi bhi logic se pehle. ChatRequest wo hai jo ek caller ko bhejna hi hoga; ChatResponse wo hai jo aapke apne handler ko bhejne ki ijaazat hai:

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

Ye Step 1 kyun hai, kisi bhi route se pehle: pydantic har incoming body ko ChatRequest ke against free mein validate karta hai -- ek request jismein message missing hai, ya jo temperature=5 bhejti hai jabki field le=2.0 kehta hai, aapki function body tak kabhi pahunchti hi nahi; FastAPI ek 422 return karta hai ek loc array ke saath jo exactly bataata hai kaunsa field galat tha, koi generic 400 nahi. Aur route ko response_model=ChatResponse (neeche use kiya gaya) se declare karna ek OUTPUT contract hai: agar aapke handler ke dict mein kabhi galti se ek extra key aa jaaye -- ek internal id, ek debug field -- pydantic use client tak pahunchne se pehle drop kar deta hai. Validation aur response shape ke liye aap if checks nahi likhte; aap unhe ek baar declare karte ho, ek type ki tarah.

Step 2 -- JWT auth ek dependency ki tarah; har route ko ek real user_id milta hai

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"]

Ek dependency kyun, har handler ke andar manual check nahi: ek baar jab route ki function signature user_id: str = Depends(current_user) padhti hai, wo handler user_id ke BINA likha hi nahi ja sakta -- code mein koi bhi path aisa nahi hai jo isse skip kare, ek haath se likhe if not token: ... ke ulat jo das alag functions ke top pe ho, jinmein se ek din koi bhool jaayega. Aur do alag except clauses matter karte hain: jwt.ExpiredSignatureError aur jwt.InvalidTokenError genuinely alag situations hain -- ek bare except: "ye token pehle valid tha aur expire ho gaya" aur "ye token forged ya corrupt hai" ko ek hi message mein collapse kar deta, aur KeyboardInterrupt bhi swallow kar leta.

Ek fact jo plainly bolna zaroori hai: ek JWT signed hota hai, encrypted nahi. Jiske paas bhi token hai wo secret key ke bina uska payload decode kar sakta hai -- secret sirf ye prove karta hai ki payload ke saath chhed-chhaad nahi hui. JWT payload mein kabhi bhi kuch aisa mat daaliye jo aap ek postcard pe print karna pasand na karo.

Kab use karein: koi bhi endpoint jise ye jaanna zaroori hai ki wo KISKA data touch kar raha hai -- jo, ek se zyada user hote hi, is project ka har endpoint hai. Kab NAHI use karein: ek genuinely public, unauthenticated endpoint (health check, public pricing page) pe Depends(current_user) bilkul nahi hona chahiye -- ek aisi route pe auth force karna jismein koi user-specific data protect karne ko nahi hai, sirf anonymous callers ke liye use todta hai, kuch fayda hue bina.

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

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

# /chat handler ke andar, is user_id ke liye `answer` compute karne ke baad:
history = chat_history.setdefault(user_id, [])
history.append({"role": "user", "content": req.message})
history.append({"role": "assistant", "content": answer})

user_id se hi keyed kyun, aur kuch nahi: dictionary ki key hi yahan POORA isolation mechanism hai. chat_history["user_alice"] aur chat_history["user_bob"] do bilkul alag lists hain sirf isliye kyunki wo do alag dict keys hain -- kahin bhi koi code aisa nahi hai jo explicitly kahe "alice ko bob ke messages mat dikhao"; ye isolation data structure ki shape ka ek side effect hai. Yehi wajah hai ki is project mein KAHIN BHI user_id include karna bhool jaana (Step 5 ka cache wo jagah hai jo log bhool jaate hain) wahi guarantee todta hai. user_id Step 2 ke Depends(current_user) se aata hai -- ye kabhi request body se nahi padha jaata, kyunki ek client body field mein kuch bhi daal sakta hai, par wo ek JWT forge nahi kar sakta jiska secret uske paas nahi hai.

Step 4 -- Model call se PEHLE ek quota check (uske baad, aap pehle hi paisa de chuke hain)

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

# handler ke andar, ISI order mein:
check_quota(user_id)      # 1. budget check karo
# ... cache lookup, phir model call sirf MISS pe hi chalti hai ...
record_usage(user_id)     # 2. sirf tab count karo jab ek answer actually serve ho gaya ho

ORDER hi poora lesson kyun hai: check_quota cache lookup se pehle aur model kabhi call hone se pehle chalta hai. Agar aap quota model call karne ke BAAD check karte, to aap pehle hi us request ke tokens spend kar chuke hote jise aap abhi refuse karne wale hain -- check ek receipt hota, ek limit nahi. Ek 429 status code (403 nahi, koi silent empty response nahi) CALLER ke code ko batata hai "ye ek rate/plan limit hai, permissions problem ya crash nahi," aur detail message ek insaan ko ek upgrade path deta hai, ek mystery failure ki jagah.

Kab use karein: koi bhi endpoint jo har call pe aapko paisa ya ek shared resource costs karta hai -- jo is project ka har route hai jo model tak pahunchta hai. Kab NAHI use karein / trade-off: ye ek product/billing concern hai, rate limiting ek SECURITY control ki tarah se alag (abusive traffic ko plan se independent throttle karna) -- wo ek doosre chapter ka kaam hai; dono ko mila mat kariye warna aap ek business rule ko ek security tool se enforce kar rahe hoge ya ulta.

Step 5 -- Ek response cache jiski key mein user id shaamil hai (isse chhod do to tenants ke across leak hota hai)

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,   # <-- isse chhod do to tenant B tenant A ka answer padh lega
        "model": model,
        "system_prompt": system_prompt,
        "message": message,
    }
    blob = json.dumps(payload, sort_keys=True)   # sort_keys: same dict, same key, field order kuch bhi ho
    return hashlib.sha256(blob.encode()).hexdigest()

Sirf message nahi, CHAAR fields kyun: message akela ek key ke liye kaafi nahi hai -- do requests identical message carry kar sakti hain aur legitimately do alag cached answers chahti hain. system_prompt key mein hona chahiye kyunki ek alag persona/instruction set badal deta hai ki sahi answer kya hai bhi; isse chhod do to ek persona ka cached reply doosre ke instructions ke neeche serve ho jaata hai. model bhi usi wajah se key mein hona chahiye -- ek sasta aur ek mehenga model same prompt ka alag jawab de sakte hain. Aur user_id wo hai jo log bhool jaate hain, kyunki ek cache "obviously" infrastructure lagta hai, user data nahi -- par is project mein actually do users ne identical message bheja, aur user_id ka hash ke andar hona hi WAHI EK wajah hai ki bob ki call ek MISS thi, na ki chupke se alice ka cached answer return kar deti (neeche output mein verified). sort_keys=True bhi matter karta hai: json.dumps iske bina bilkul same dict ke liye ek alag string (aur isliye ek alag hash) produce kar sakta hai, field insertion order pe depend karte hue.

Kab use karein: repeated, non-personalised, time-sensitive nahi requests, jahan same answer ke liye do baar pay karna pure waste hai -- yahan ek cache HIT kuch sau microseconds mein measure hota hai ek model call ke against jo (ek real provider ke saath) sau-sau milliseconds ya zyada leti hai. Kab NAHI use karein: ek genuinely personalised answer, koi bhi time-sensitive cheez (aaj ki date, "pichhli baar poochne ke baad kya badla"), ya koi bhi request jahan temperature 0 se upar set hai SPECIFICALLY isliye kyunki caller variety chahta hai -- wahan caching paisa nahi bachati, chupke se feature todti hai.

Step 6 -- 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"},
    )

Stream kyun karein bilkul -- aur iske baare mein kabhi na bolne wali baat: streaming ek faster TIME-TO-FIRST-TOKEN deti hai -- caller pehla word lagbhag turant dekh leta hai, poora answer generate hone ka wait kiye bina. Ye answer ko overall faster nahi banati -- total generation time usually same ya thoda worse hota hai ek single blocking call se; interview mein ulta mat bolna.

"Bas streaming add kar do" ka non-obvious half: HTTP status code (200) aur headers client ko usi pal bhej diye jaate hain jab PEHLA chunk jaata hai -- upar verified, response data: frames stream karna shuru kar deta hai answer ka ek bhi word exist karne se pehle. Agar 10 mein se 5th word generate karte waqt kuch fail ho jaaye, aap wapas jaakar us 200 ko 500 mein nahi badal sakte -- status line pehle hi ja chuki hai. event_generator ke andar ka except block decoration nahi hai; ye stream shuru hone ke baad ek error report karne ki EK BAAKI jagah hai, ek aur data: frame ki tarah jise frontend ko check karna padega.

Kab use karein: server se ek single caller ki taraf tokens ka one-way flow, jo ek chat reply hoti hai. Kab NAHI use karein / trade-off: SSE one-way hai (server -> client) aur plain HTTP pe chalti hai; agar aapko CLIENT ko bhi usi connection pe messages wapas push karne hain (ek live collaborative editor, bidirectional presence), wo ek WebSocket ka kaam hai, SSE ka nahi -- ek one-way token stream ke liye WebSocket mat pakdo, ye reconnect aur scaling complexity add karta hai jo yahan zaroori nahi.

Step 7 -- Document upload ek background job ki tarah + poll karne ke liye ek status endpoint

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)   # parsing/chunking/embedding ke liye stand-in
    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

Response kaam khatam hone se pehle kyun return hota hai: background_tasks.add_task(...) process_document ko HTTP response bhej diye jaane ke BAAD chalne ke liye schedule karta hai -- neeche verified: POST /documents 9 ms mein wapas aa gaya jabki neeche asli 1-second wala job abhi bhi chal raha tha. Ek poore second ke liye (ya real document ke liye lambe waqt tak) request ko khula rakhna exactly wo hai jo proxy timeouts cause karta hai aur agar caller ka connection drop ho jaaye to kaam kho deta hai -- accept-then-poll dono se bachta hai. GET /documents/{job_id} usi in-memory dict ka ek plain read hai, jise caller status ke "done" banne tak poll karta hai.

Ye teen limits jo isse solve NAHI hoti (bina poochhe ye bolo): jobs ek plain dict hai EK process ki RAM mein -- ek second worker process jo ek alag request serve kar raha hai, use bilkul nahi pata hoga ki ye job exist karti hai; koi retry nahi hai, to agar process_document fail ho jaaye, job chupke se hamesha ke liye "running" rehti hai bina kahin bhi koi error surface hue; aur koi concurrency cap nahi hai, to 100 simultaneous uploads 100 background tasks start kar dete hain bina kisi throttle ke. Ye teeno gaps exactly wo hain jo ek real task queue (Celery, RQ, ek managed cloud queue) close karne ke liye banaayi jaati hai -- ye project jaan-boojh kar wahin ruk jaata hai jahan se aapko ek queue chahiye hogi, aur use describe karta hai fake-run karne ki jagah, kyunki in mein se koi bhi package yahan installed nahi hai.

Kab use karein: kuch bhi jo ek request se zyada slow hai jise reasonably wait karna chahiye -- document ingestion, batch scoring, report generation. Kab NAHI use karein / trade-off: kuch bhi jiska jawab CALLER ko turant chahiye (khud /chat endpoint) ko background job hone ka koi matlab nahi -- accept-then-poll ek turant jawab ko ek delayed jawab se trade karta hai, jo primary chat path ke liye galat hai.

Step 8 -- Tokens, latency aur cache status ki structured logging, aur ek 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}},
    }

Ek structured dict log karo, ek sentence nahi, kyun: print(f"user {user_id} used {tokens} tokens") readable hai par QUERYABLE nahi -- aap sentences ke ek dher se "kal mera cache hit rate kya tha" nahi poochh sakte. Ek structured entry (json.dumps(entry)) wo hai jise ek real logging/observability stack (ye project sirf print karta hai; ek real system ek log aggregator ko bhejta hai) filter, group aur aggregate kar sakta hai. Upar ka /admin/tally endpoint hi wo aggregation hai, in-memory list pe haath se ki hui: is project ki apni run mein verified, chaar logged requests ne ek real cache_hit_rate 0.25 diya (chaar mein se ek HIT) aur ek real per-user breakdown -- yehi shape hai ek resume line ki jo kehti hai "measured a 25% cache hit rate" na ki "added caching."

Jahan ek real provider call fake_llm ki jagah leti hai (illustrative)

# ---- ILLUSTRATIVE (requires an API key; not executed) ----
# Upar har jagah, fake_llm(req.message) iski jagah stand-in hai. Ise swap karne se
# auth, quota, cache key, streaming, ya background job mein kuch nahi badalta -- sirf
# ek function ka body hai, bilkul unhi jagahon se call hota hai.
def real_llm(message: str, system_prompt: str) -> str:
    response = provider_client.chat.completions.create(
        model="<current-small-chat-model>",  # model IDs badalte rehte hain -- provider ki current list check karo
        messages=[{"role": "system", "content": system_prompt},
                  {"role": "user", "content": message}],
    )
    return response.choices[0].message.content

Interview mein ise kaise bolna hai (honest version)

Ye project genuinely kya prove karta hai: aap request path ke har station ko bana sakte hain jo ek real AI backend ko chahiye -- auth (ek JWT dependency jispe har route depend karta hai), identity-scoped state (per-user history AND per-user quota AND ek cache key jismein user id shaamil hai, teeno verified ki users ke across leak nahi hote), call se pehle cost control (quota check hota hai model chalne se pehle, baad mein nahi), ek streaming endpoint jo ek blocking response ki jagah Server-Sent Events wapas deta hai, request path se hataya gaya kaam (ek background job jiska status poll kar sakte ho, measured 9 ms mein instead of ek second block hone ke), aur structured logs jo ek tally feed karte hain (tokens, latency, cache status, per-user counts). Is project ke output mein har number ek real chalte hue server se aaya hai, kisi description se nahi.

Ye abhi kya NAHI karta, aur interviewer ke dhoondhne se pehle aapko khud bolna chahiye:

• Single process, in-memory stores. chat_history, usage_counts, response_cache aur jobs sab plain Python dicts hain jo is EK process ki RAM mein rehte hain. Process restart karo aur har user ki history, quota counter, cache aur job status sab chala jaata hai. Ek second worker process (jo real traffic ke liye chahiye hota) bhi in mein se kuch nahi dekhega -- har worker ki apni copy hoti hai.
• Koi real queue nahi. BackgroundTasks usi web process mein chalta hai. Agar process_document fail ho jaaye to koi retry nahi hai, agar 100 uploads ek saath aa jaayein to koi concurrency limit nahi hai, aur cross-worker visibility bhi nahi hai. Yehi exactly wo cheez hai jo ek real task queue (Celery, RQ, ek managed cloud queue) fix karne ke liye banayi gayi hai -- ye project jaan-boojh kar wahin ruk jaata hai jahan se aapko ek queue chahiye hogi.
• Model call illustrative hai, chalayi nahi gayi. fake_llm / fake_llm_stream deterministic stand-ins hain taaki poora request path -- auth, quota, cache, streaming, background jobs, logging -- zero API keys ke saath genuinely chale. Ek real provider call swap karne se upar ki wiring mein kuch nahi badalta; sirf ek function ka body badalta hai.
• Koi database nahi. Ek real product ko history, quota counters aur cache ko restart ke baad bhi survive karna chahiye -- iska matlab ek relational database ya Redis, dict nahi. Ye project har ek ki SHAPE sikhata hai (key kya hai, kab check hota hai, usmein kya hona chahiye); use persist karna agla step hai, ye nahi.

Ye limits khud, bina poochhe, naam se bolna hi wo cheez hai jo "maine ek chatbot banaya" ko "maine multi-user backend ke parts banaye aur mujhe exactly pata hai ye kahan production-ready hona band karta hai" se alag karti hai.

Standard definition (interview me bolo): 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}},
    }