## Why This Project Exists The RAG chatbot from Project 1 works on your laptop. That is not the same as being a service other people, or other systems, can rely on. Without authentication, anyone who finds the URL can query it. Without rate limiting, one misbehaving client can exhaust your API budget in minutes. Without tracing, a slow response is a mystery, was it retrieval, reranking, or generation that took 4 seconds? Without caching, you pay full LLM cost for the same question asked twice in a row. This project takes the RAG chatbot (or the refund agent from Project 2, either works) and wraps it in the infrastructure that makes it an actual production service. ### What you are building * A FastAPI service with Pydantic request/response validation and streaming responses * API key authentication and per-key rate limiting * Distributed tracing following a request through retrieval, reranking, and generation * Response and semantic caching to cut cost on repeated or near-duplicate queries * Containerisation with Docker and deployment to a managed cloud AI service ### Architecture diagram Client Request (with API key) | v +----------------+ | Auth + Rate | <-- rejects invalid/over-limit here | Limit Check | +----------------+ | v +----------------+ +------------------+ | Check Cache | --> | Cache Hit: Return | +----------------+ +------------------+ | Miss v +----------------+ | RAG Pipeline | <-- each stage emits a trace span | (Project 1) | +----------------+ | v +----------------+ | Store in Cache| +----------------+ | v +----------------+ | Stream Response| +----------------+ > 📌 **Engineering Decision:** A managed model API for almost everything, self-hosting only once data residency requires it or API cost at real traffic volume exceeds GPU cost. This project deploys to a managed cloud AI service for exactly that reason, the operational cost of running your own inference infrastructure is rarely justified before you have both the traffic and the constraint that demands it. ### Prerequisites This project assumes you have completed Project 1 (Production RAG Chatbot) or Project 2 (Reliable AI Agent), you will be wrapping that existing pipeline, not building a new one from scratch. > 💡 **Tip:** If you completed both Project 1 and Project 2, wrap the RAG chatbot here, its streaming response behaviour makes the FastAPI streaming milestone more concrete to test.
## Why FastAPI and Not a Plain Script A Python script that calls your RAG pipeline works for one person on one laptop. A service other systems can call needs a defined request/response contract, input validation before your expensive pipeline ever runs, and the ability to handle many requests concurrently. FastAPI with Pydantic gives you all three with comparatively little code. ### Defining the request and response contract ```python from fastapi import FastAPI, HTTPException from pydantic import BaseModel, Field app = FastAPI(title="HR Policy Chatbot API") class ChatRequest(BaseModel): question: str = Field(..., min_length=3, max_length=500) employee_id: str = Field(..., description="Used for audit logging, not personalization") class ChatResponse(BaseModel): answer: str sources: list[str] latency_ms: float ``` > **Note:** `min_length=3, max_length=500` on the question field rejects a malformed or empty request before it reaches the RAG pipeline at all. Validating input at the API boundary, not deep inside your pipeline code, means a bad request fails fast and cheaply instead of burning an embedding call and an LLM call on garbage input. ### The main endpoint ```python import time @app.post("/chat", response_model=ChatResponse) async def chat(request: ChatRequest): """ Main chat endpoint. Pydantic has already validated request.question and request.employee_id by the time this function body runs. """ start = time.time() try: candidates = hybrid_search(request.question, db_conn, top_k=10) reranked = rerank(request.question, candidates, top_n=4) answer = generate_answer(request.question, reranked) except Exception as e: raise HTTPException(status_code=500, detail="Failed to generate answer") latency = (time.time() - start) * 1000 return ChatResponse( answer=answer, sources=[c["source"] for c in reranked], latency_ms=round(latency, 2) ) ``` ### Adding streaming responses A 3-second wait with no feedback feels broken to a user, even when the service is working correctly. Streaming tokens as they generate fixes the perceived latency without changing the actual generation time. ```python from fastapi.responses import StreamingResponse @app.post("/chat/stream") async def chat_stream(request: ChatRequest): candidates = hybrid_search(request.question, db_conn, top_k=10) reranked = rerank(request.question, candidates, top_n=4) async def token_generator(): # Assumes generate_answer_streaming yields tokens as the LLM produces them async for token in generate_answer_streaming(request.question, reranked): yield token return StreamingResponse(token_generator(), media_type="text/event-stream") ``` > 🔴 **Common Mistake:** No timeout handling on long-running calls, especially streaming responses. A client that disconnects mid-stream, or an LLM call that hangs, can leave a request open indefinitely on the server, consuming resources for a response nobody will ever receive. Wrap generation calls in an explicit timeout and handle client disconnection. ### Guided practice Run the service locally with `uvicorn main:app --reload` and send both a normal `/chat` request and a `/chat/stream` request using `curl`. Confirm the streaming endpoint visibly returns tokens incrementally rather than all at once.
## Why This Matters More Once Actions Are Involved An unauthenticated read-only endpoint is a mild risk. An endpoint that can trigger a refund agent taking real financial actions, as in Project 2, without authentication is a serious one. Every service in this project's scope needs to know who is calling it and how often, before it does any real work. ### API key authentication ```python from fastapi import Security, HTTPException from fastapi.security import APIKeyHeader api_key_header = APIKeyHeader(name="X-API-Key") ## In production, load from a secrets manager, never hardcode VALID_API_KEYS = { "key_rahul_hr_team": {"tenant": "hr_team", "rate_limit": 100}, "key_priya_support": {"tenant": "support_team", "rate_limit": 50} } async def verify_api_key(api_key: str = Security(api_key_header)): if api_key not in VALID_API_KEYS: raise HTTPException(status_code=401, detail="Invalid API key") return VALID_API_KEYS[api_key] ``` ```python @app.post("/chat", response_model=ChatResponse) async def chat(request: ChatRequest, key_info: dict = Security(verify_api_key)): ## key_info now available for tenant isolation and rate limit lookups ... ``` > ⚠️ **Security:** Never hardcode API keys directly in source code, even for a project like this one. Load them from environment variables or a secrets manager, and confirm `.env` or equivalent secret files are listed in `.gitignore` before your first commit. ### Rate limiting per API key ```python from collections import defaultdict from datetime import datetime, timedelta request_log = defaultdict(list) # in production, use Redis instead of in-memory def check_rate_limit(api_key: str, limit_per_minute: int): """ Sliding window rate limit, counts requests in the last 60 seconds for this specific key and rejects if the limit is exceeded. """ now = datetime.utcnow() window_start = now - timedelta(seconds=60) request_log[api_key] = [ t for t in request_log[api_key] if t > window_start ] if len(request_log[api_key]) >= limit_per_minute: raise HTTPException(status_code=429, detail="Rate limit exceeded") request_log[api_key].append(now) ``` > **Note:** This in-memory implementation resets every time the service restarts and does not work correctly across multiple server instances behind a load balancer. It is fine for local development and this project's lab, a real production deployment needs a shared store like Redis so every instance sees the same rate limit counters. ### Tenant isolation ```python @app.post("/chat", response_model=ChatResponse) async def chat(request: ChatRequest, key_info: dict = Security(verify_api_key)): check_rate_limit(key_info["tenant"], key_info["rate_limit"]) ## Filter retrieval to only this tenant's documents, a support-team ## key should never be able to retrieve HR-team-only policy chunks candidates = hybrid_search( request.question, db_conn, top_k=10, tenant=key_info["tenant"] ) ... ``` > 🔴 **Common Mistake:** Treating authentication and tenant isolation as the same thing. Authentication confirms who is calling, tenant isolation confirms what that caller is allowed to see or act on. A valid API key alone does not mean the caller should see every document or trigger every action in the system. ### Troubleshooting scenario A client reports getting `429 Too Many Requests` after only 3 calls, well under their configured limit of 100. Check whether `check_rate_limit` is being called with the correct key, a bug where every request accidentally shares one global counter key instead of a per-tenant one would produce exactly this symptom under concurrent testing.
## Why Logs Alone Are Not Enough A log line saying "request took 4.2 seconds" tells you the total time, not where it went. Was retrieval slow because the vector index needs rebuilding? Was reranking slow because too many candidates were passed in? Was generation slow because of a large context window? **Tracing** breaks a single request into named spans, each with its own duration, answering exactly that question. ### Instrumenting the pipeline with spans ```python from opentelemetry import trace from opentelemetry.trace import Status, StatusCode tracer = trace.get_tracer("rag_chatbot_service") async def traced_chat_pipeline(question: str, tenant: str): with tracer.start_as_current_span("chat_request") as root_span: root_span.set_attribute("tenant", tenant) with tracer.start_as_current_span("hybrid_retrieval") as span: candidates = hybrid_search(question, db_conn, top_k=10, tenant=tenant) span.set_attribute("candidates_found", len(candidates)) with tracer.start_as_current_span("reranking") as span: reranked = rerank(question, candidates, top_n=4) span.set_attribute("top_score", reranked[0]["rerank_score"]) with tracer.start_as_current_span("generation") as span: answer = generate_answer(question, reranked) span.set_attribute("answer_length", len(answer)) return answer, reranked ``` > **Note:** A span is a named, timed segment of work with its own start and end. Nesting spans (retrieval and reranking and generation all inside the parent `chat_request` span) is what lets a tracing dashboard show a request as a timeline, each stage's actual duration, instead of one opaque total number. ### Attaching evaluation scores to traces ```python with tracer.start_as_current_span("faithfulness_check") as span: context = "\n".join(c["content"] for c in reranked) faithfulness_result = score_faithfulness(context, answer) span.set_attribute("faithfulness_verdict", faithfulness_result) ``` > 📌 **Engineering Decision:** Attach evaluation scores directly to traces, not just to a separate offline evaluation report. When a specific production request produces a bad answer, being able to click into that exact trace and see its faithfulness score, retrieval candidates, and generation latency together is what turns "users are complaining" into a specific, fixable root cause. ### Cost attribution ```python with tracer.start_as_current_span("generation") as span: response = client.messages.create(...) span.set_attribute("input_tokens", response.usage.input_tokens) span.set_attribute("output_tokens", response.usage.output_tokens) span.set_attribute("tenant", tenant) # Cost per request, per tenant, now queryable from trace data directly ``` > 🔴 **Common Mistake:** No request logging or tracing, meaning nobody can answer "why did this cost so much" after the fact. Without token counts attached to traces, a sudden spike in API spend is a mystery you can only investigate by guessing, with tracing it is a query you can run directly. ### Guided practice Send 10 requests through the traced pipeline with varying question lengths, then inspect the resulting traces and confirm you can identify which stage, retrieval, reranking, or generation, consumed the most time for the slowest of the 10.
## Why Caching Is a Compact but Real Cost Lever Support chatbots see the same handful of questions repeatedly: "how many leave days," "what is the WFH policy." Paying full embedding, retrieval, reranking, and generation cost for a question asked an hour ago by a different employee is waste that caching eliminates almost for free. ### Exact-match response caching ```python import hashlib import redis cache = redis.Redis(host="localhost", port=6379, decode_responses=True) def get_cache_key(question: str, tenant: str) -> str: normalized = question.strip().lower() return f"chat_cache:{tenant}:{hashlib.sha256(normalized.encode()).hexdigest()}" def get_cached_response(question: str, tenant: str): key = get_cache_key(question, tenant) cached = cache.get(key) return json.loads(cached) if cached else None def set_cached_response(question: str, tenant: str, response: dict, ttl_seconds=3600): key = get_cache_key(question, tenant) cache.setex(key, ttl_seconds, json.dumps(response)) ``` > **Note:** `ttl_seconds=3600` means a cached answer expires after one hour. This matters because HR policies change, a cache with no expiry could serve a stale answer about a leave policy that was updated yesterday. The TTL is a deliberate tradeoff between cost savings and freshness, not an arbitrary number. ### Semantic caching for near-duplicate questions Exact-match caching misses "how many casual leaves do I get" and "how many casual leave days are there" even though they mean the same thing. Semantic caching checks embedding similarity against recently cached questions instead of exact string matching. ```python def get_semantic_cached_response(question: str, tenant: str, threshold=0.95): """ Check if a highly similar question was answered recently. threshold=0.95 is intentionally strict, a looser threshold risks returning a cached answer to a question that only sounds similar but actually needs a different answer. """ question_embedding = embedder.encode(question) recent_cached = get_recent_cache_entries(tenant, limit=50) for entry in recent_cached: similarity = cosine_similarity(question_embedding, entry["embedding"]) if similarity >= threshold: return entry["response"] return None ``` > 🔴 **Common Mistake:** No caching on repeated or near-duplicate requests, wasting spend on identical work performed over and over. A support chatbot answering the same top-5 questions hundreds of times a day without caching pays full pipeline cost every single time, when a one-hour TTL cache would eliminate the vast majority of that repeated cost. ### Wiring caching into the endpoint ```python @app.post("/chat", response_model=ChatResponse) async def chat(request: ChatRequest, key_info: dict = Security(verify_api_key)): check_rate_limit(key_info["tenant"], key_info["rate_limit"]) cached = get_cached_response(request.question, key_info["tenant"]) if cached: return ChatResponse(**cached, latency_ms=2.0) # cache hits are near-instant answer, reranked = await traced_chat_pipeline(request.question, key_info["tenant"]) response = {"answer": answer, "sources": [c["source"] for c in reranked]} set_cached_response(request.question, key_info["tenant"], response) return ChatResponse(**response, latency_ms=...) ``` ### Concept check Before continuing, explain in one sentence why the semantic cache threshold is set to 0.95 rather than something looser like 0.80. Consider what happens to answer correctness if two genuinely different questions are treated as cache-equivalent.
## Why Docker and a Managed Path, Not a Full Cloud Checklist The goal here is one clean, working deployment path, not a tour of every cloud AI product available. Containerising your service in Docker makes it portable, and deploying to a managed platform means you are not managing GPU infrastructure yourself for a workload that does not need it. ### Writing the Dockerfile ```dockerfile FROM python:3.11-slim WORKDIR /app ## Copy requirements first so Docker caches this layer ## when only application code changes, not dependencies COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . EXPOSE 8000 CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"] ``` > 💡 **Tip:** Copying `requirements.txt` and running `pip install` before copying the rest of the application code is deliberate layer ordering. Docker caches each instruction as a layer, changing only your Python source will not force a full dependency reinstall on the next build, changing `requirements.txt` will, which is the correct behaviour either way. ### Building and testing locally ```bash ## Build the image, tagged for this specific service docker build -t hr-chatbot-service:latest . ## Run it locally with environment variables for secrets docker run -p 8000:8000 \ -e ANTHROPIC_API_KEY=$ANTHROPIC_API_KEY \ -e REDIS_HOST=host.docker.internal \ hr-chatbot-service:latest ## Confirm it responds correctly before deploying anywhere curl -X POST http://localhost:8000/chat \ -H "X-API-Key: key_rahul_hr_team" \ -H "Content-Type: application/json" \ -d '{"question": "How many casual leave days do I get?", "employee_id": "EMP-4021"}' ``` > 🔴 **Common Mistake:** Unpinned dependencies breaking a working local setup in production. If `requirements.txt` specifies `fastapi` with no version pin, a build six months from now may pull a newer version with breaking changes, even though nothing in your code changed. Pin exact versions for anything deployed. ### Deploying to a managed cloud AI service ```bash ## Example path: push the built image to a container registry, ## then deploy it to a managed container service in ap-south-1 (Mumbai) ## for lower latency to Indian users docker tag hr-chatbot-service:latest <registry-url>/hr-chatbot-service:latest docker push <registry-url>/hr-chatbot-service:latest ## Deploy using your cloud provider's managed container run command, ## setting the region explicitly and secrets via the provider's ## secret manager rather than plain environment variables in the console ``` > ⚠️ **Security:** Never pass API keys or database credentials as plain environment variables typed directly into a cloud console. Use the provider's secrets manager and reference secrets by name in your deployment configuration, so the actual values never appear in deployment logs or console history. ### Documenting the cost difference After deploying, compare running this service locally versus on the managed platform for a simulated 1000 requests a day. Record: managed platform compute cost, LLM API cost (unchanged either way), and the operational time saved not managing servers yourself. This comparison is the deliverable the capstone project spec asks for. ### Capstone lab checkpoint Deploy the containerised service to your chosen managed cloud AI platform. Run the same `curl` test from local testing against the deployed URL and confirm identical behaviour. Write a one-page cost report comparing local, self-managed cloud VM, and managed platform costs at your simulated traffic volume.
Why This Project Exists The RAG chatbot from Project 1 works on your laptop. That is not the same as being a service oth...
Why FastAPI and Not a Plain Script A Python script that calls your RAG pipeline works for one person on one laptop. A se...
Why This Matters More Once Actions Are Involved An unauthenticated read-only endpoint is a mild risk. An endpoint that c...
Why Logs Alone Are Not Enough A log line saying "request took 4.2 seconds" tells you the total time, not where it went. ...
Why Caching Is a Compact but Real Cost Lever Support chatbots see the same handful of questions repeatedly: "how many le...
Why Docker and a Managed Path, Not a Full Cloud Checklist The goal here is one clean, working deployment path, not a tou...
Final Verification 1. Authentication check Expected output: 401 Unauthorized, not a successful response. 2. Rate limit c...
Aligns directly with DevOps, Site Reliability (SRE), and Platform Engineering job descriptions.