Ep 98: Building Your Own AI SaaS: The Complete Architecture

You've made it. 98 episodes. Tokens to transformers. Training to inference. Embeddings to voice assistants.

Now let's put it all together.

Building an AI SaaS isn't just calling the OpenAI API. It's auth, billing, rate limiting, model routing, observability, scaling, and the 3 AM page when your GPU cluster runs out of memory.

This is the architecture you need. Everything we've learned, in one system.

The Architecture Overview

┌─────────────────────────────────────────────────────────────────────┐

│ CLIENT LAYER │

│ React/Vue App → React Native → API Clients (Python, JS, etc) │

└─────────────────────────────────────────────────────────────────────┘

example
code
↓

│ GATEWAY LAYER │

│ Auth (JWT) → Rate Limit → API Gateway → Router │

example
code
↓

│ SERVICE LAYER │

│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │

│ │ Auth │ │ Billing │ │ Request │ │ Admin │ │

│ │ Service │ │ Service │ │ Service │ │ Service │ │

│ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │

example
code
↓

│ AI ORCHESTRATION │

│ Cache (Semantic) → Model Router → Fallback Handler │

│ ↓ │

│ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ │

│ │ OpenAI │ │ Claude │ │ Self- │ │ Other │ │

│ │ │ │ │ │ Hosted │ │ Providers │

│ └─────────┘ └─────────┘ └─────────┘ └─────────┘ │

example
code
↓

│ DATA & OBSERVABILITY │

│ Postgres (Users) → Redis (Cache) → ClickHouse (Analytics) │

│ ↓ ↓ ↓ │

│ S3 (Logs/Exports) Vector DB Grafana/Datadog │

⚡

Component 1: Authentication & Authorization

See Episode 54 for why auth matters for data privacy.

JWT-Based Auth

python

from fastapi import FastAPI, Depends, HTTPException

from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

import jwt

snippet
code
app = FastAPI()
security = HTTPBearer()
async def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
try:
        payload = jwt.decode(credentials.credentials, SECRET_KEY, algorithms=["HS256"])
        return payload
    except jwt.InvalidTokenError:
        raise HTTPException(401, "Invalid token")

@app.post("/v1/chat")

snippet
code
async def chat(request: ChatRequest, user: dict = Depends(verify_token)):
# user["sub"] = user_id
    # user["plan"] = subscription tier
    pass

API Key Management

python

import secrets

snippet
code
def generate_api_key():
return f"sk-{secrets.token_urlsafe(32)}"

Store hashed version in DB

snippet
code
api_key_hash = hashlib.sha256(api_key.encode()).hexdigest()

🔧

Component 2: Rate Limiting & Quotas

See Episode 84 for gateway patterns.

Tiered Rate Limits

python

PLAN_LIMITS = {

example
code
"free": {"rpm": 10, "tpm": 1000, "daily_cost": 0.10},
    "pro": {"rpm": 100, "tpm": 10000, "daily_cost": 5.00},
    "enterprise": {"rpm": 1000, "tpm": 100000, "daily_cost": 100.00}

}

async def check_rate_limit(user_id: str, plan: str):

example
code
limits = PLAN_LIMITS[plan]

example
code
# Check requests per minute
    current_rpm = await redis.get(f"rpm:{user_id}")
    if int(current_rpm or 0) >= limits["rpm"]:
        raise HTTPException(429, "Rate limit exceeded")

example
code
# Check daily spend
    daily_spend = await redis.get(f"spend:{user_id}:{today()}")
    if float(daily_spend or 0) >= limits["daily_cost"]:
        raise HTTPException(429, "Daily quota exceeded")

Token Bucket in Redis

lua

-- rate_limit.lua

local key = KEYS[1]

local limit = tonumber(ARGV[1])

local window = tonumber(ARGV[2])

local current = redis.call("GET", key)

if not current then

example
code
redis.call("SET", key, 1, "EX", window)
    return 1

end

if tonumber(current) < limit then

example
code
redis.call("INCR", key)
    return tonumber(current) + 1

else

example
code
return -1

end

📊

Component 3: The AI Gateway

This is the heart of your system. See Episode 84.

Unified API

python

from fastapi import FastAPI

import litellm

snippet
code
app = FastAPI()

@app.post("/v1/chat/completions")

async def chat_completions(request: ChatCompletionRequest):

example
code
# 1. Check cache (Episode 83)
    cached = await check_semantic_cache(request.messages)
    if cached:
        return {"choices": [{"message": {"content": cached}}], "cached": True}

example
code
# 2. Route to appropriate model
    model = route_model(request.messages, request.model)

example
code
# 3. Call with fallback
    try:
        response = await call_with_fallback(model, request)
    except Exception as e:
        raise HTTPException(503, f"All providers failed: {e}")

example
code
# 4. Store in cache
    await store_in_semantic_cache(request.messages, response)

example
code
# 5. Log usage
    await log_usage(request.user_id, model, response.usage)

example
code
return response

Semantic Caching

See Episode 83.

python

from sentence_transformers import SentenceTransformer

import qdrant_client

snippet
code
embedder = SentenceTransformer('all-MiniLM-L6-v2')
qdrant = qdrant_client.QdrantClient(host="localhost", port=6333)
async def check_semantic_cache(messages: list, threshold=0.92):
query = messages[-1]["content"]  # Last user message
    embedding = embedder.encode(query)
results = qdrant.search(
        collection_name="response_cache",
        query_vector=embedding.tolist(),
        limit=1
    )
if results and results[0].score > threshold:
        return results[0].payload["response"]
    return None

Component 4: Model Routing

See Episode 55 for cost optimization strategies.

Smart Router

python

snippet
code
def route_model(messages: list, requested_model: str = None) -> str:
# User specified
    if requested_model:
        return requested_model
# Auto-route based on content
    content = messages[-1]["content"]
# Simple tasks → Cheap model
    if is_simple_task(content):
        return "claude-3.5-haiku"
# Code tasks → Best coding model
    if is_code_task(content):
        return "claude-3.5-sonnet"
# Long context → Gemini
    if estimate_tokens(messages) > 100000:
        return "gemini-1.5-pro"
# Default
    return "gpt-4-turbo"
def is_simple_task(text: str) -> bool:
simple_keywords = ["what is", "define", "explain simple"]
    return any(kw in text.lower() for kw in simple_keywords)

💡

Component 5: Billing System

Track every token, bill accurately.

Usage Tracking

python

from decimal import Decimal

async def log_usage(user_id: str, model: str, usage: dict):

example
code
cost = calculate_cost(model, usage["prompt_tokens"], usage["completion_tokens"])

example
code
# Store in ClickHouse for analytics
    await clickhouse.insert("usage_logs", {
        "user_id": user_id,
        "model": model,
        "input_tokens": usage["prompt_tokens"],
        "output_tokens": usage["completion_tokens"],
        "cost_usd": cost,
        "timestamp": datetime.utcnow()
    })

example
code
# Update Redis for real-time quota
    await redis.incrbyfloat(f"spend:{user_id}:{today()}", cost)

MODEL_PRICING = {

example
code
"gpt-4": {"input": 30/1_000_000, "output": 60/1_000_000},
    "claude-3.5-haiku": {"input": 0.25/1_000_000, "output": 1.25/1_000_000},

}

snippet
code
def calculate_cost(model: str, input_tokens: int, output_tokens: int) -> Decimal:
pricing = MODEL_PRICING.get(model, {"input": 0, "output": 0})
    return Decimal(
        input_tokens * pricing["input"] +
        output_tokens * pricing["output"]
    )

🔬

Component 6: Observability

See Episode 56 for full observability setup.

Structured Logging

python

import structlog

snippet
code
logger = structlog.get_logger()

async def log_request(request_id: str, user_id: str, model: str, latency: float):

example
code
logger.info(
        "ai_request_completed",
        request_id=request_id,
        user_id=user_id,
        model=model,
        latency_ms=latency,
        environment="production"
    )

Metrics

python

from prometheus_client import Counter, Histogram, Gauge

snippet
code
requests_total = Counter('ai_requests_total', 'Total requests', ['model', 'status'])
request_duration = Histogram('ai_request_duration_seconds', 'Request duration', ['model'])
active_requests = Gauge('ai_active_requests', 'Active requests')

@app.post("/v1/chat/completions")

async def chat(request: Request):

example
code
active_requests.inc()
    start = time.time()

example
code
try:
        response = await process(request)
        requests_total.labels(model=request.model, status="success").inc()
        return response
    except Exception as e:
        requests_total.labels(model=request.model, status="error").inc()
        raise
    finally:
        request_duration.labels(model=request.model).observe(time.time() - start)
        active_requests.dec()

🛡️

Component 7: Scaling Infrastructure

Horizontal Scaling

yaml

docker-compose.yml

version: '3.8'

services:

api:

example
code
build: .
    replicas: 3
    environment:
      - REDIS_URL=redis://redis:6379
      - DATABASE_URL=postgresql://...

redis:

example
code
image: redis:7-alpine

postgres:

example
code
image: postgres:15
    volumes:
      - postgres_data:/var/lib/postgresql/data

Kubernetes Deployment

yaml

k8s-deployment.yaml

apiVersion: apps/v1

kind: Deployment

metadata:

name: ai-api

spec:

replicas: 5

selector:

example
code
matchLabels:
      app: ai-api

template:

example
code
spec:
      containers:
      - name: api
        image: your-registry/ai-api:latest
        resources:
          requests:
            memory: "512Mi"
            cpu: "500m"
          limits:
            memory: "1Gi"
            cpu: "1000m"
        env:
        - name: OPENAI_API_KEY
          valueFrom:
            secretKeyRef:
              name: api-secrets
              key: openai-key

📦

Component 8: Security & Compliance

See Episode 54 for privacy considerations.

Data Handling

python

PII detection before sending to LLM

from presidio_analyzer import AnalyzerEngine

from presidio_anonymizer import AnonymizerEngine

snippet
code
analyzer = AnalyzerEngine()
anonymizer = AnonymizerEngine()
def sanitize_prompt(text: str) -> str:
results = analyzer.analyze(text=text, language='en')
    if results:
        anonymized = anonymizer.anonymize(text=text, analyzer_results=results)
        return anonymized.text
    return text

Request Validation

python

from pydantic import BaseModel, validator

class ChatRequest(BaseModel):

example
code
messages: list
    model: str
    max_tokens: int = 150

example
code
@validator('max_tokens')
    def limit_tokens(cls, v):
        if v > 4096:
            raise ValueError('max_tokens cannot exceed 4096')
        return v

example
code
@validator('messages')
    def limit_context(cls, v):
        if len(v) > 50:
            raise ValueError('Too many messages in context')
        return v

🚀

The Complete Request Flow

Client sends POST /v1/chat/completions

example
code
↓

JWT verification

example
code
↓

Rate limit check (Redis)

example
code
↓

Semantic cache check (Qdrant)

example
code
↓ [cache miss]

Model routing

example
code
↓

Request validation (PII, limits)

example
code
↓

Call LLM API (with fallback)

example
code
↓

Store in semantic cache

example
code
↓

Log usage (ClickHouse)

example
code
↓

Metrics (Prometheus)

example
code
↓

Return response

🔄

Cost Structure at Scale

Component Monthly Cost (1M requests) Notes

LLM APIs $10,000-50,000 Biggest cost by far

Compute (API servers)	$500-2,000	3-5 instances
Redis (cache)	$200-500	Managed Redis
Postgres	$100-300	User data
Vector DB	$300-800	Semantic cache
Observability	$200-500	Datadog/Grafana
Total	$11,300-54,100	Varies by model mix

🏗️

What We've Built

This architecture combines everything from the series:

snippet
code
Tokens (Episode 1) — We count and limit them

Training vs Inference (Episode 4) — We only do inference

snippet
code
Embeddings (Episode 15) — We cache by meaning

Vector Search (Episode 17) — Fast semantic cache lookup

Cost Optimization (Episode 55) — Smart routing, caching

snippet
code
Monitoring (Episode 56) — Full observability

Semantic Caching (Episode 83) — 40-70% cost savings

AI Gateway (Episode 84) — Unified API

Data Privacy (Episode 54) — PII handling, compliance