Deploy an AI Service with FastAPI, Auth, Tracing, and Caching

Wrap the RAG chatbot in a production FastAPI service with auth, rate limiting, tracing, caching, and cloud deployment.

Related Concepts & TermsDeployment

Domains & Technologies

Domains

FASTAPIOBSERVABILITYLLMOPS

Technologies

DOCKER

Blueprint Walkthrough

Architecture Overview & Problem Statement

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

◈ 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.

Milestone 1: Serve the Pipeline with FastAPI and Pydantic

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.

Milestone 2: Add API Authentication and Rate Limiting

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.

Milestone 3: Add Distributed Tracing Through the Pipeline

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.

Milestone 4: Add Response and Semantic Caching

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.

Milestone 5: Containerise and Deploy to a Managed Cloud AI Service

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.

Validation & Testing

Final Verification

1. Authentication check

Bash

## Request with no API key should be rejected
curl -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"question": "test", "employee_id": "EMP-1"}'

Expected output: 401 Unauthorized, not a successful response.

2. Rate limit check

Send requests rapidly exceeding your configured per-minute limit for one API key. Expected output: requests beyond the limit return 429 Too Many Requests, and a different API key's requests are unaffected in the same window.

3. Tracing check

Send one request and inspect the resulting trace. Expected output: separate spans for hybrid_retrieval, reranking, and generation, each with its own duration, summing to approximately the total request latency.

4. Cache check

Send the identical question twice in a row. Expected output: the second request's latency_ms is dramatically lower than the first, confirming the cache hit path executed instead of the full pipeline.

5. Container check

Bash

docker run -p 8000:8000 hr-chatbot-service:latest
curl http://localhost:8000/docs

Expected output: the FastAPI auto-generated docs page loads successfully from inside the container, confirming the image is self-contained and correctly configured.

Quick reference

Layer	Purpose	Failure symptom if missing
Pydantic validation	Reject bad input early	Garbage reaches the expensive pipeline
API key auth	Know who is calling	Anyone can use your service and budget
Rate limiting	Prevent abuse	One client exhausts shared capacity
Tracing	Localize slowness	"It's slow somehow" with no fix path
Caching	Cut repeated cost	Paying full price for the same question twice

Common mistakes across the full project

No request logging or tracing leaves you unable to answer why a bill spiked after the fact, and the fix is attaching token counts and tenant identity to every trace from day one. No timeout or streaming handling on long calls leaves users staring at nothing during a slow response, and the fix is explicit timeouts plus streaming for anything that can take more than a second or two. Unpinned dependencies breaking a working local setup in production happens when version numbers are left open-ended in requirements.txt, and the fix is pinning exact versions for anything that gets deployed. Self-hosting for cost reasons without calculating real GPU cost at actual traffic volume often costs more than the managed API it was meant to save money over, and the fix is running the numbers in Milestone 5's cost report before committing to either path. No caching on repeated or near-duplicate requests wastes spend that a simple one-hour TTL cache would eliminate almost entirely, and the fix is treating caching as a default, not an afterthought bolted on after a high cost bill arrives.

Tip
This deployed service is a real, working example of the full production architecture pattern taught conceptually in the LLMOps module, revisit this project after that module to see how each concept there maps to code you already wrote here.

Videos & Guides

DOCUMENTATION

FastAPI Official Documentation

Official FastAPI docs covering request validation, streaming responses, and security.

View Reference

DOCUMENTATION

OpenTelemetry Python Documentation

Reference for instrumenting Python services with distributed tracing spans.

View Reference