Deploy a FastAPI AI Service with Auth and Tracing

Wrap a RAG chatbot in a production FastAPI service with API key auth, rate limiting, OpenTelemetry tracing, Redis caching and Docker deployment.

Architecture Overview & Problem Statement

This project turns a working RAG chatbot into a production FastAPI service. You add Pydantic validation, streaming responses, API key authentication, per-key rate limiting in Redis, OpenTelemetry tracing across retrieval, reranking and generation, exact and semantic caching, and a Docker image deployed to a managed container platform. By the end, other systems can call your pipeline safely, you can see where every second and every token went, and repeated questions stop costing you money.

A RAG pipeline that runs on your laptop is not yet a service. With no authentication, anyone who finds the URL can query it. With no rate limit, one misbehaving client can burn through your LLM budget in minutes. With no tracing, a slow answer is a mystery: was it retrieval, reranking or generation? With no cache, you pay full price for a question that an hour-ago colleague already asked. Each layer in this project closes one of those gaps.

flowchart TD client([Client request]):::code --> auth[Auth and
rate limit]:::tool auth --> cache{Cache hit?}:::tool cache -->|hit| reply([Return answer]):::good cache -->|miss| rag[RAG pipeline
traced spans]:::cloud rag --> store[(Store in Redis)]:::data store --> reply classDef code fill:#DBEAFE,color:#1E293B classDef tool fill:#475569,color:#FFFFFF classDef cloud fill:#8B5CF6,color:#FFFFFF classDef data fill:#F97316,color:#FFFFFF classDef good fill:#10B981,color:#FFFFFF

What you will build

A FastAPI service with Pydantic request and response models and a streaming endpoint
API key authentication, per-key rate limiting in Redis, and tenant isolation
OpenTelemetry traces with one span per pipeline stage, plus token counts per tenant
An exact-match cache and a semantic cache with a TTL and an index version
A pinned, non-root Docker image deployed to a managed platform, with a written cost comparison

Prerequisites

You need the pipeline from the Production RAG Chatbot project (or the refund agent from the Reliable AI Agent project). This project wraps that code and does not rebuild it. The snippets assume your pipeline exposes hybrid_search, rerank, generate_answer and a streaming variant generate_answer_streaming, plus a sentence-embedding model called embedder. You also need Docker, Python 3.11 and a Redis instance (a local container is fine).

Tip
If you finished both earlier projects, wrap the RAG chatbot. Its token streaming makes the streaming milestone much easier to see and test.

Engineering Decision
Use a managed model API and a managed container platform for almost everything. Self-host inference only when data residency demands it or when your measured API bill at real traffic exceeds the cost of running GPUs. Milestone 5 makes you compute that comparison with your own numbers.

Milestone 1: Serve the Pipeline with FastAPI and Pydantic

This milestone exposes your RAG pipeline as a validated HTTP API with a normal endpoint and a streaming endpoint.

A script works for one person on one laptop. A service needs a fixed request and response contract, validation before the expensive pipeline runs, and concurrent request handling. FastAPI with Pydantic gives you all three, and generates interactive docs at /docs for free.

Create this layout:

Bash

mkdir -p rag-service/app && cd rag-service
touch app/__init__.py app/main.py app/models.py app/pipeline.py requirements.txt

Create app/models.py:

Python

from pydantic import BaseModel, Field
 
 
class ChatRequest(BaseModel):
    question: str = Field(..., min_length=3, max_length=500)
    employee_id: str = Field(..., description="Used for audit logging, not personalisation")
 
 
class ChatResponse(BaseModel):
    answer: str
    sources: list[str]
    latency_ms: float
    cache: str = "miss"  # "miss", "exact" or "semantic"

Input limits on question reject empty and oversized requests at the edge. A bad request then fails in microseconds instead of spending an embedding call and an LLM call on garbage.

Create app/pipeline.py as the single place that calls your existing RAG code. Keeping it separate means the web layer never needs to know how retrieval works:

Python

from your_rag_package import (
    hybrid_search,
    rerank,
    generate_answer,
    db_conn,
    embedder,
)
 
 
def run_pipeline(question: str, tenant: str) -> tuple[str, list[dict]]:
    candidates = hybrid_search(question, db_conn, top_k=10, tenant=tenant)
    reranked = rerank(question, candidates, top_n=4)
    answer = generate_answer(question, reranked)
    return answer, reranked

Now the first version of the app. Your pipeline functions are blocking, so the endpoint hands them to a thread pool. Calling them directly inside async def would freeze the event loop and serialise every request.

Create app/main.py:

Python

import asyncio
import time
 
from fastapi import FastAPI, HTTPException
from fastapi.concurrency import run_in_threadpool
 
from app.models import ChatRequest, ChatResponse
from app.pipeline import run_pipeline
 
app = FastAPI(title="HR Policy Chatbot API")
REQUEST_TIMEOUT_S = 25
 
 
@app.get("/healthz")
def healthz():
    return {"status": "ok"}
 
 
@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    start = time.perf_counter()
    try:
        async with asyncio.timeout(REQUEST_TIMEOUT_S):
            answer, reranked = await run_in_threadpool(
                run_pipeline, request.question, "hr_team"
            )
    except TimeoutError:
        raise HTTPException(status_code=504, detail="Answer timed out")
    except Exception:
        raise HTTPException(status_code=500, detail="Failed to generate answer")
 
    return ChatResponse(
        answer=answer,
        sources=[c["source"] for c in reranked],
        latency_ms=round((time.perf_counter() - start) * 1000, 2),
    )

Add a streaming endpoint

A three-second silence feels broken even when the service is fine. Streaming tokens as they are produced fixes perceived latency without changing real generation time. The browser-standard format is Server-Sent Events, where each message is data: ... followed by a blank line.

Add to app/main.py:

Python

from fastapi import Request
from fastapi.responses import StreamingResponse
from your_rag_package import generate_answer_streaming
from app.pipeline import hybrid_search, rerank, db_conn
 
 
@app.post("/chat/stream")
async def chat_stream(request: ChatRequest, http_request: Request):
    candidates = await run_in_threadpool(hybrid_search, request.question, db_conn, 10, "hr_team")
    reranked = await run_in_threadpool(rerank, request.question, candidates, 4)
 
    async def event_stream():
        try:
            async with asyncio.timeout(REQUEST_TIMEOUT_S):
                async for token in generate_answer_streaming(request.question, reranked):
                    if await http_request.is_disconnected():
                        break  # stop paying for tokens nobody will read
                    yield f"data: {token}\n\n"
            yield "event: done\ndata: [DONE]\n\n"
        except TimeoutError:
            yield "event: error\ndata: timeout\n\n"
 
    return StreamingResponse(event_stream(), media_type="text/event-stream")

Common Mistake
Having no timeout and no disconnect check on streaming calls. A client that drops mid-stream, or an LLM call that hangs, keeps the request open and keeps spending tokens for a reader who left. Wrap generation in a timeout and stop when is_disconnected() is true, as above.

Run and test it

Bash

pip install fastapi "uvicorn[standard]" pydantic
uvicorn app.main:app --reload

In a second terminal, compare the two endpoints. The -N flag turns off curl buffering so you can see tokens arrive:

Bash

curl -s -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"question": "How many casual leave days do I get?", "employee_id": "EMP-4021"}'
 
curl -N -X POST http://localhost:8000/chat/stream \
  -H "Content-Type: application/json" \
  -d '{"question": "How many casual leave days do I get?", "employee_id": "EMP-4021"}'

Expected output: the first command prints one JSON body after a pause, and the second prints data: lines one after another as tokens are generated.

Tip
The streaming endpoint skips response_model, so Pydantic does not check what you send. Keep the request model strict, since that is where bad input arrives.

Milestone 2: Add API Key Authentication and Rate Limiting

This milestone makes every request prove who it is, caps how often each key can call you, and keeps each tenant inside its own documents.

An unauthenticated read-only endpoint is a mild risk. An endpoint that can trigger actions, like the refund agent from the earlier project, is a serious one. Authentication answers who is calling. Tenant isolation answers what that caller may see. They are different checks, and a valid key alone must never mean access to every document.

Store key hashes, load them from the environment

Never keep raw keys in source code. Keep only SHA-256 hashes, loaded from an environment variable or a secrets manager. A leaked config then does not reveal usable keys.

Generate a key and its hash:

Bash

python -c "
import secrets, hashlib
key = 'sk_' + secrets.token_urlsafe(32)
print('KEY (give to client):', key)
print('HASH (store server-side):', hashlib.sha256(key.encode()).hexdigest())
"

Put the hashes in API_KEYS_JSON, for example (one line in real use):

JSON

{
  "HASH_OF_HR_KEY": {"tenant": "hr_team", "rate_limit": 100},
  "HASH_OF_SUPPORT_KEY": {"tenant": "support_team", "rate_limit": 50}
}

Create app/security.py:

Python

import hashlib
import json
import os
import time
import uuid
 
import redis
from fastapi import HTTPException, Security
from fastapi.security import APIKeyHeader
 
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)
API_KEYS = json.loads(os.environ["API_KEYS_JSON"])
redis_client = redis.Redis.from_url(
    os.environ.get("REDIS_URL", "redis://localhost:6379/0"), decode_responses=True
)
 
 
async def verify_api_key(api_key: str | None = Security(api_key_header)) -> dict:
    if not api_key:
        raise HTTPException(status_code=401, detail="Missing API key")
    digest = hashlib.sha256(api_key.encode()).hexdigest()
    info = API_KEYS.get(digest)
    if info is None:
        raise HTTPException(status_code=401, detail="Invalid API key")
    return {**info, "key_id": digest[:12]}
 
 
def check_rate_limit(key_id: str, limit_per_minute: int) -> None:
    """Sliding window in Redis: count this key's requests in the last 60 seconds."""
    now = time.time()
    rl_key = f"rl:{key_id}"
    pipe = redis_client.pipeline()
    pipe.zremrangebyscore(rl_key, 0, now - 60)
    pipe.zadd(rl_key, {f"{now}:{uuid.uuid4().hex}": now})
    pipe.zcard(rl_key)
    pipe.expire(rl_key, 61)
    _, _, count, _ = pipe.execute()
    if count > limit_per_minute:
        raise HTTPException(
            status_code=429,
            detail="Rate limit exceeded",
            headers={"Retry-After": "60"},
        )

Using auto_error=False lets you return a clean 401 for a missing key. The default behaviour can answer with 403 depending on your FastAPI version, which confuses clients.

The counter lives in Redis, so every server instance behind a load balancer sees the same numbers, and a restart does not reset anyone's window. Requests that get rejected still count toward the window, which gently punishes a client that hammers a 429.

Protect both endpoints and isolate tenants

Update app/main.py so the tenant comes from the key, never from the request body:

Python

from fastapi import Security
from app.security import verify_api_key, check_rate_limit
 
 
@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest, key_info: dict = Security(verify_api_key)):
    check_rate_limit(key_info["key_id"], key_info["rate_limit"])
    tenant = key_info["tenant"]
    ...  # call run_pipeline(request.question, tenant) as before

Apply the same two dependencies to /chat/stream, and pass tenant into hybrid_search so the database filter only returns that tenant's chunks. A support-team key must never retrieve HR-only policy text.

Security
Add .env and any file holding keys to .gitignore before your first commit. In deployment, inject API_KEYS_JSON from a secrets manager. If a key leaks, remove its hash and redeploy, since the hash cannot be turned back into the key.

Common Mistake
Treating authentication and tenant isolation as the same thing. A valid key proves identity only. Always filter retrieval and cache lookups by the tenant tied to that key.

Troubleshooting a surprise 429

A client gets 429 after only 3 calls, far below their limit of 100. Check which value you pass into check_rate_limit. If every request accidentally uses one shared string as the key, all clients share a single counter and the limit trips under concurrent traffic. The first argument must be the per-key key_id.

Milestone 3: Add Distributed Tracing Through the Pipeline

This milestone makes each request show up as a timeline with one named span per stage, so you can see where time and tokens go.

A log line saying "request took 4.2 seconds" gives the total and nothing else. Tracing splits one request into nested spans, each with a start, an end and attributes. A tracing UI then draws the request as a timeline, which turns "it is slow somehow" into "reranking took 2.8 seconds because 40 candidates were passed in".

flowchart TD root[chat_request]:::root --> a[hybrid_retrieval]:::step root --> b[reranking]:::step root --> c[generation]:::step root --> d[faithfulness_check]:::opt classDef root fill:#8B5CF6,color:#FFFFFF classDef step fill:#DBEAFE,color:#1E293B classDef opt fill:#FFEDD5,color:#1E293B

Set up OpenTelemetry

Bash

pip install opentelemetry-sdk opentelemetry-exporter-otlp-proto-http \
  opentelemetry-instrumentation-fastapi

Create app/tracing.py:

Python

from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
 
 
def setup_tracing(app):
    provider = TracerProvider(resource=Resource.create({"service.name": "rag-chatbot-service"}))
    # The endpoint comes from OTEL_EXPORTER_OTLP_ENDPOINT, default http://localhost:4318
    provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
    trace.set_tracer_provider(provider)
    FastAPIInstrumentor.instrument_app(app)

Call setup_tracing(app) once in app/main.py after creating the app. Start a local trace viewer. Pin the image tag once you pick a version:

Bash

docker run -d --name jaeger -p 16686:16686 -p 4318:4318 jaegertracing/all-in-one:latest

Wrap each stage in a span

Update app/pipeline.py. Because this function runs in a worker thread and FastAPI already started a request span, your spans nest under it automatically:

Python

import random
 
from opentelemetry import trace
 
tracer = trace.get_tracer("rag_chatbot_service")
FAITHFULNESS_SAMPLE_RATE = 0.1
 
 
def run_pipeline(question: str, tenant: str):
    with tracer.start_as_current_span("chat_request") as root:
        root.set_attribute("tenant", tenant)
        root.set_attribute("question_length", len(question))  # length only, never the text
 
        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"] if reranked else 0.0)
 
        with tracer.start_as_current_span("generation") as span:
            answer, usage = generate_answer(question, reranked, return_usage=True)
            span.set_attribute("input_tokens", usage.input_tokens)
            span.set_attribute("output_tokens", usage.output_tokens)
            span.set_attribute("tenant", tenant)
 
        if random.random() < FAITHFULNESS_SAMPLE_RATE:
            with tracer.start_as_current_span("faithfulness_check") as span:
                context = "\n".join(c["content"] for c in reranked)
                span.set_attribute("faithfulness_verdict", str(score_faithfulness(context, answer)))
 
        return answer, reranked

This snippet assumes your Project 1 generate_answer can return the model's usage object (add a return_usage flag that passes back response.usage) and that score_faithfulness is imported from your evaluation code. Three choices here are deliberate:

Token counts and tenant sit on the generation span, so "which tenant caused the cost spike" becomes a trace query instead of a guess.
The faithfulness check runs on a sample of requests, because the judge is itself an LLM call and checking every request would add cost and latency to every answer.
Spans record question_length, not the question. Employee questions can contain personal data, and trace backends are widely readable.

Engineering Decision
Attach evaluation scores to traces, not only to an offline report. When one production answer is bad, you can open that exact trace and see its retrieval candidates, rerank score, faithfulness verdict and generation time together.

Common Mistake
Shipping with no tracing or token logging. When the bill jumps, nobody can say which tenant, endpoint or prompt caused it. Add token counts and tenant to every generation span from the first release.

Guided practice

Send 10 requests with varied question lengths, then open Jaeger at http://localhost:16686. Pick the slowest trace and name the stage that used the most time.

Milestone 4: Add Response and Semantic Caching

This milestone stops you paying full pipeline cost for the same question twice, using an exact-match cache first and a semantic cache for rephrased questions.

Support chatbots see the same handful of questions all day, such as leave days or the work-from-home policy. A cache hit skips embedding, retrieval, reranking and generation entirely.

Exact-match cache with a TTL and an index version

Two details prevent stale answers. The TTL makes entries expire, so a changed policy is not served forever. The index version is part of the key, so re-indexing your documents instantly makes every old cached answer unreachable.

Create app/cache.py:

Python

import hashlib
import json
import os
 
import numpy as np
 
from app.security import redis_client
 
INDEX_VERSION = os.environ.get("INDEX_VERSION", "1")
TTL_SECONDS = 3600
SEMANTIC_THRESHOLD = 0.95
SEMANTIC_MAX_ENTRIES = 50
 
 
def _exact_key(question: str, tenant: str) -> str:
    normalized = " ".join(question.lower().split())
    digest = hashlib.sha256(normalized.encode()).hexdigest()
    return f"chat:{tenant}:v{INDEX_VERSION}:{digest}"
 
 
def get_exact(question: str, tenant: str):
    cached = redis_client.get(_exact_key(question, tenant))
    return json.loads(cached) if cached else None
 
 
def set_exact(question: str, tenant: str, response: dict):
    redis_client.setex(_exact_key(question, tenant), TTL_SECONDS, json.dumps(response))

Semantic cache for near-duplicate questions

Exact matching misses "how many casual leaves do I get" versus "how many casual leave days are there". A semantic cache compares embeddings of the new question against recent cached ones. With normalised embeddings, a dot product equals cosine similarity.

Add to app/cache.py:

Python

def _sem_key(tenant: str) -> str:
    return f"sem:{tenant}:v{INDEX_VERSION}"
 
 
def get_semantic(embedding: np.ndarray, tenant: str):
    for raw in redis_client.lrange(_sem_key(tenant), 0, SEMANTIC_MAX_ENTRIES - 1):
        entry = json.loads(raw)
        if float(np.dot(embedding, np.array(entry["embedding"]))) >= SEMANTIC_THRESHOLD:
            return entry["response"]
    return None
 
 
def set_semantic(embedding: np.ndarray, tenant: str, response: dict):
    key = _sem_key(tenant)
    redis_client.lpush(key, json.dumps({"embedding": embedding.tolist(), "response": response}))
    redis_client.ltrim(key, 0, SEMANTIC_MAX_ENTRIES - 1)
    redis_client.expire(key, TTL_SECONDS)

The threshold is strict on purpose. Set too loose, it returns a cached answer for a question that only sounds similar but needs a different answer. "How many casual leaves do I get" and "How many sick leaves do I get" can score very high on some models and still have different answers.

Common Mistake
Picking a semantic threshold without testing it. Embedding models differ, so 0.95 is a starting point, not a rule. Build a list of paraphrase pairs and a list of near-miss pairs from your own HR policies, score both, and choose the threshold that separates them.

Wire both caches into the endpoint

Update /chat in app/main.py. The order matters: authenticate, rate limit, exact cache, semantic cache, then the pipeline:

Python

from app import cache
from app.pipeline import embedder
 
 
@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest, key_info: dict = Security(verify_api_key)):
    start = time.perf_counter()
    check_rate_limit(key_info["key_id"], key_info["rate_limit"])
    tenant = key_info["tenant"]
 
    def elapsed() -> float:
        return round((time.perf_counter() - start) * 1000, 2)
 
    hit = cache.get_exact(request.question, tenant)
    if hit:
        return ChatResponse(**hit, latency_ms=elapsed(), cache="exact")
 
    embedding = await run_in_threadpool(
        embedder.encode, request.question, normalize_embeddings=True
    )
    hit = cache.get_semantic(embedding, tenant)
    if hit:
        return ChatResponse(**hit, latency_ms=elapsed(), cache="semantic")
 
    try:
        async with asyncio.timeout(REQUEST_TIMEOUT_S):
            answer, reranked = await run_in_threadpool(run_pipeline, request.question, tenant)
    except TimeoutError:
        raise HTTPException(status_code=504, detail="Answer timed out")
 
    response = {"answer": answer, "sources": [c["source"] for c in reranked]}
    cache.set_exact(request.question, tenant, response)
    cache.set_semantic(embedding, tenant, response)
    return ChatResponse(**response, latency_ms=elapsed(), cache="miss")

The tenant sits in every cache key, so one tenant can never receive another tenant's cached answer. Latency is now measured, not hard-coded, and the cache field tells you which path served each request.

Concept check

In one sentence, explain why the semantic threshold is 0.95 and not 0.80. Think about what happens to correctness when two genuinely different questions are treated as the same.

Milestone 5: Containerise and Deploy to a Managed Platform

This milestone packages the service as a pinned, non-root Docker image and deploys it to a managed container platform, with secrets kept out of the console.

Docker makes the service portable. A managed platform means you do not run servers or GPUs for a workload that does not need them. This lab shows AWS App Runner in ap-south-1 (Mumbai) because it needs the least setup. The same image runs on Google Cloud Run or Azure Container Apps with different deploy commands.

Pin your dependencies

Unpinned versions mean a rebuild next month can pull breaking changes with no code change on your side. Freeze exact versions once everything works:

Bash

pip freeze > requirements.txt

Write the Dockerfile

Create Dockerfile. Copying requirements.txt before the code keeps the dependency layer cached when only source files change:

Dockerfile

FROM python:3.11-slim
 
ENV PYTHONDONTWRITEBYTECODE=1 \
    PYTHONUNBUFFERED=1
 
WORKDIR /app
 
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
 
COPY app ./app
 
RUN useradd --create-home appuser
USER appuser
 
EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

Create .dockerignore so secrets and junk never enter the image:

Bash

.env
.git
__pycache__
*.pyc
venv

Build and run locally

Bash

docker build -t hr-chatbot-service:1.0.0 .
 
docker run -p 8000:8000 \
  -e ANTHROPIC_API_KEY=$ANTHROPIC_API_KEY \
  -e API_KEYS_JSON="$API_KEYS_JSON" \
  -e REDIS_URL=redis://host.docker.internal:6379/0 \
  hr-chatbot-service:1.0.0

Test it the same way you will test the deployed copy:

Bash

curl -X POST http://localhost:8000/chat \
  -H "X-API-Key: YOUR_RAW_KEY" \
  -H "Content-Type: application/json" \
  -d '{"question": "How many casual leave days do I get?", "employee_id": "EMP-4021"}'

Common Mistake
Tagging images latest and leaving dependencies unpinned. You cannot tell what is running or roll back to a known build. Tag releases with a version or commit SHA and pin requirements.txt.

Push the image and store secrets

Bash

aws ecr create-repository --repository-name hr-chatbot-service --region ap-south-1
 
ACCOUNT_ID=$(aws sts get-caller-identity --query Account --output text)
REGISTRY=$ACCOUNT_ID.dkr.ecr.ap-south-1.amazonaws.com
 
aws ecr get-login-password --region ap-south-1 \
  | docker login --username AWS --password-stdin $REGISTRY
 
docker tag hr-chatbot-service:1.0.0 $REGISTRY/hr-chatbot-service:1.0.0
docker push $REGISTRY/hr-chatbot-service:1.0.0
 
aws secretsmanager create-secret --name hrbot/anthropic-api-key \
  --secret-string "$ANTHROPIC_API_KEY" --region ap-south-1
aws secretsmanager create-secret --name hrbot/api-keys-json \
  --secret-string "$API_KEYS_JSON" --region ap-south-1
aws secretsmanager create-secret --name hrbot/redis-url \
  --secret-string "$MANAGED_REDIS_URL" --region ap-south-1

Create the service

Create apprunner.json. Secrets are referenced by ARN, so their values never appear in the console or in deployment history:

JSON

{
  "ServiceName": "hr-chatbot-api",
  "SourceConfiguration": {
    "AuthenticationConfiguration": {
      "AccessRoleArn": "arn:aws:iam::ACCOUNT_ID:role/AppRunnerECRAccessRole"
    },
    "AutoDeploymentsEnabled": false,
    "ImageRepository": {
      "ImageIdentifier": "ACCOUNT_ID.dkr.ecr.ap-south-1.amazonaws.com/hr-chatbot-service:1.0.0",
      "ImageRepositoryType": "ECR",
      "ImageConfiguration": {
        "Port": "8000",
        "RuntimeEnvironmentVariables": {
          "INDEX_VERSION": "1",
          "OTEL_EXPORTER_OTLP_ENDPOINT": "https://YOUR_TRACE_BACKEND"
        },
        "RuntimeEnvironmentSecrets": {
          "ANTHROPIC_API_KEY": "arn:aws:secretsmanager:ap-south-1:ACCOUNT_ID:secret:hrbot/anthropic-api-key-XXXXXX",
          "API_KEYS_JSON": "arn:aws:secretsmanager:ap-south-1:ACCOUNT_ID:secret:hrbot/api-keys-json-XXXXXX",
          "REDIS_URL": "arn:aws:secretsmanager:ap-south-1:ACCOUNT_ID:secret:hrbot/redis-url-XXXXXX"
        }
      }
    }
  },
  "InstanceConfiguration": {
    "Cpu": "1 vCPU",
    "Memory": "2 GB",
    "InstanceRoleArn": "arn:aws:iam::ACCOUNT_ID:role/hrbot-instance-role"
  },
  "HealthCheckConfiguration": {
    "Protocol": "HTTP",
    "Path": "/healthz"
  }
}

Bash

aws apprunner create-service --cli-input-json file://apprunner.json --region ap-south-1

The instance role needs secretsmanager:GetSecretValue on those three secrets and nothing else. Your Redis must be reachable from the service. A hosted Redis with a TLS URL works for the lab, and ElastiCache needs a VPC connector.

Security
Never type API keys or database URLs as plain environment variables in a cloud console. Use the platform's secrets manager and reference the secret by name or ARN, so values stay out of console history and deploy logs.

Document the cost difference

Run a simulated 1,000 requests a day for a few days locally and on the managed platform. Record your own measured numbers:

Item	Local or laptop	Self-managed cloud VM	Managed platform
Compute cost	Fill in	Fill in	Fill in
LLM API cost (from trace token counts)	Fill in	Fill in	Fill in
Cache hit rate	Fill in	Fill in	Fill in
Operations time per week	Fill in	Fill in	Fill in

Use your traces for the LLM line: sum input_tokens and output_tokens per tenant and multiply by your provider's current prices. Write a one-page report that says which option you would choose at this traffic and what traffic or constraint would change your answer.

Capstone lab checkpoint

Deploy the container, run the same curl test against the deployed URL, and confirm the response and status codes match local behaviour.

Validation & Testing

These checks confirm each layer works and fails safely. Replace BASE with http://localhost:8000 locally or your deployed URL.

1. Authentication

Bash

curl -s -o /dev/null -w "%{http_code}\n" -X POST $BASE/chat \
  -H "Content-Type: application/json" \
  -d '{"question": "test question", "employee_id": "EMP-1"}'

Expected output: 401. Repeat with a wrong key and expect 401 again. A correct key returns 200.

2. Input validation

Bash

curl -s -o /dev/null -w "%{http_code}\n" -X POST $BASE/chat \
  -H "X-API-Key: YOUR_RAW_KEY" -H "Content-Type: application/json" \
  -d '{"question": "hi", "employee_id": "EMP-1"}'

Expected output: 422, with no embedding or LLM call made.

3. Rate limiting and key isolation

Bash

for i in $(seq 1 60); do
  curl -s -o /dev/null -w "%{http_code} " -X POST $BASE/chat \
    -H "X-API-Key: SUPPORT_KEY" -H "Content-Type: application/json" \
    -d "{\"question\": \"rate test number $i\", \"employee_id\": \"EMP-1\"}"
done; echo

Expected output: with the support key limited to 50 per minute, the first 50 return 200 and the rest return 429. A request with the HR key in the same minute still returns 200.

4. Tenant isolation

Ask a question that only HR documents can answer using the support key.

Expected output: the answer does not contain HR-only content, and the sources list holds only support-team documents.

5. Tracing

Send one uncached request and open its trace in your viewer.

Expected output: spans hybrid_retrieval, reranking and generation nest under the request, and their durations add up to roughly the total latency. The generation span shows input_tokens, output_tokens and tenant.

6. Caching

Send the same question twice, then a rephrased version.

Expected output: the first response shows "cache": "miss" with a normal latency. The identical repeat shows "cache": "exact" with a far lower latency_ms. The rephrase shows "cache": "semantic" only if its similarity clears your threshold. Also confirm a near-miss question (different leave type) shows "miss".

7. Streaming

Bash

curl -N -X POST $BASE/chat/stream \
  -H "X-API-Key: YOUR_RAW_KEY" -H "Content-Type: application/json" \
  -d '{"question": "Explain the work from home policy", "employee_id": "EMP-1"}'

Expected output: data: lines appear gradually and the stream ends with a done event. Press Ctrl+C mid-stream and confirm the server logs stop generating.

8. Container

Bash

docker run --rm -p 8000:8000 --env-file .env hr-chatbot-service:1.0.0
curl -s http://localhost:8000/healthz

Expected output: {"status":"ok"}, and http://localhost:8000/docs loads from inside the container.

Quick reference

Layer	Purpose	Symptom if missing
Pydantic validation	Reject bad input early	Garbage reaches the expensive pipeline
API key auth	Know who is calling	Anyone can spend your budget
Redis rate limit	Stop abuse across instances	One client exhausts shared capacity
Tenant isolation	Limit what each caller sees	A valid key reads every document
Tracing	Locate slowness and cost	"It is slow somehow" with no fix path
Caching with TTL and version	Cut repeated cost safely	Full price for the same question, or stale answers
Timeouts and disconnect checks	Free resources from dead requests	Hung requests and wasted tokens

Portfolio deliverable

Keep the repository, a screenshot of one trace with all three spans, your threshold test results, and the one-page cost report. Record your measured cache hit rate and cost figures, since they depend on your traffic and prices.

Found a mistake, an outdated step or a broken link?

Deploy a FastAPI AI Service with Auth and Tracing

Blueprint Walkthrough

Architecture Overview & Problem Statement

What you will build

Prerequisites

Milestone 1: Serve the Pipeline with FastAPI and Pydantic

Add a streaming endpoint

Run and test it

Milestone 2: Add API Key Authentication and Rate Limiting

Store key hashes, load them from the environment

Protect both endpoints and isolate tenants

Troubleshooting a surprise 429

Milestone 3: Add Distributed Tracing Through the Pipeline

Set up OpenTelemetry

Wrap each stage in a span

Guided practice

Milestone 4: Add Response and Semantic Caching

Exact-match cache with a TTL and an index version

Semantic cache for near-duplicate questions

Wire both caches into the endpoint

Concept check

Milestone 5: Containerise and Deploy to a Managed Platform

Pin your dependencies

Write the Dockerfile

Build and run locally

Push the image and store secrets

Create the service

Document the cost difference

Capstone lab checkpoint

Validation & Testing

1. Authentication

2. Input validation

3. Rate limiting and key isolation

4. Tenant isolation

5. Tracing

6. Caching

7. Streaming

8. Container

Quick reference

Portfolio deliverable