7. Deployment and Optimization

Description

You have the individual components: DocumentProcessor, VisionAnalyzer, RAGModule, AudioModule. They work on your local machine. Now you need them to work in production: packaged in Docker, exposed as a REST API with FastAPI, with secure environment variables, structured logging, rate limiting, cost monitoring, and latency optimizations. This capsule turns your local project into a deployable service.

Why it matters: A system that only works on your laptop isn't portfolio-worthy. The difference between "I made a project" and "I built a service" is deployment: Docker, REST API, HTTP error handling, logging, health checks, and externalized configuration. These are the patterns a hiring manager looks for.

Connection with the module: In Module 7 you learned production patterns: use-case selection, fallbacks, cost optimization. Here you apply them to the complete Document Analyzer: each pattern materializes into code, configuration, or infrastructure.

Production Checklist

Infrastructure

• Multi-stage Dockerfile for an optimized build
• docker-compose.yml with service + volumes
• Environment variables from .env (not hardcoded)
• API keys as secrets, never in code
• Functional health check endpoint
• Structured logging in JSON

REST API

• Input validation (type, size, format)
• Rate limiting per IP (10 requests/minute)
• Configurable timeout per operation
• Error handling with appropriate HTTP codes
• CORS configured for the frontend (if applicable)
• Automatic documentation with Swagger/OpenAPI

Optimization

• Cache of image descriptions (don't re-describe)
• Cache of generated audio (don't re-synthesize)
• Use gpt-4o-mini where it's enough
• Reduce image resolution before Vision
• Persistent RAG index (don't re-index every request)
• Background processing for large documents

Monitoring

• Metrics: latency per operation, errors, estimated costs
• Logging of each request with duration and result
• Tracking of accumulated costs per API key
• Alerts on error threshold (>5% in 5 min)

FastAPI: Complete Endpoints

Application structure

import logging
import os
import time
import uuid
from contextlib import asynccontextmanager
from pathlib import Path
from typing import Optional

from fastapi import FastAPI, File, Form, HTTPException, UploadFile, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse, JSONResponse
from pydantic import BaseModel, Field

from modules.document_processor import DocumentProcessor
from modules.vision_analyzer import VisionAnalyzer
from modules.rag_module import RAGModule
from modules.audio_module import AudioModule
from models.schemas import AnalyzeResponse, QARequest, QAResult

logger = logging.getLogger(__name__)


@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.processor = DocumentProcessor()
    app.state.analyzer = VisionAnalyzer()
    app.state.rag = RAGModule(persist_directory="./chroma_data")
    app.state.audio = AudioModule(output_dir="./audio_output")
    app.state.start_time = time.time()
    logger.info("Services initialized")
    yield
    logger.info("Services closed")


app = FastAPI(
    title="Document Analyzer Multimodal",
    description="API for multimodal document analysis",
    version="1.0.0",
    lifespan=lifespan
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)


UPLOAD_DIR = "/tmp/uploads"
MAX_FILE_SIZE = int(os.getenv("MAX_FILE_SIZE_MB", "50")) * 1024 * 1024
SUPPORTED_EXTENSIONS = {".pdf", ".png", ".jpg", ".jpeg", ".webp"}

Endpoint POST /analyze

@app.post("/analyze", response_model=AnalyzeResponse)
async def analyze_document(
    request: Request,
    file: UploadFile = File(...),
    question: Optional[str] = Form(None),
    extract_structured: bool = Form(True),
    generate_summary: bool = Form(True),
    generate_audio_summary: bool = Form(False),
    index_for_qa: bool = Form(True),
    audio_voice: str = Form("nova")
):
    start = time.time()
    doc_id = str(uuid.uuid4())
    errors: list[str] = []

    ext = Path(file.filename).suffix.lower()
    if ext not in SUPPORTED_EXTENSIONS:
        raise HTTPException(400, f"Format '{ext}' not supported")

    os.makedirs(UPLOAD_DIR, exist_ok=True)
    file_path = os.path.join(UPLOAD_DIR, f"{doc_id}{ext}")

    try:
        content_bytes = await file.read()
        if len(content_bytes) > MAX_FILE_SIZE:
            raise HTTPException(413, "File exceeds the maximum size")

        with open(file_path, "wb") as f:
            f.write(content_bytes)

        processor = request.app.state.processor
        analyzer = request.app.state.analyzer
        rag = request.app.state.rag
        audio = request.app.state.audio

        processed = processor.process(file_path)

        doc_type = "other"
        try:
            doc_type = analyzer.classify(processed)
        except Exception as e:
            errors.append(f"Classification failed: {e}")
            logger.warning(f"Classification failed for {doc_id}: {e}")

        extracted_data = None
        if extract_structured:
            try:
                extraction = analyzer.extract_structured(processed, doc_type)
                extracted_data = {
                    "document_type": extraction.document_type,
                    "fields": extraction.fields,
                    "confidence": extraction.confidence
                }
            except Exception as e:
                errors.append(f"Structured extraction failed: {e}")
                logger.warning(f"Extraction failed for {doc_id}: {e}")

        summary = None
        if generate_summary:
            try:
                summary = _generate_summary(analyzer, processed)
            except Exception as e:
                errors.append(f"Summary failed: {e}")

        indexed = False
        if index_for_qa:
            try:
                image_descriptions = None
                if processed.has_image_pages:
                    image_descriptions = analyzer.describe_for_rag(
                        processed.get_images_for_vision()
                    )
                rag.index(
                    doc_id=doc_id,
                    content=processed,
                    filename=file.filename,
                    document_type=doc_type,
                    image_descriptions=image_descriptions
                )
                indexed = True
            except Exception as e:
                errors.append(f"Indexing failed: {e}")

        qa_result = None
        if question:
            try:
                if indexed:
                    qa = rag.query(question=question, doc_id=doc_id)
                    qa_result = {
                        "question": qa.question,
                        "answer": qa.answer,
                        "sources": qa.sources,
                        "confidence": qa.confidence
                    }
                else:
                    qa_result = _direct_qa(analyzer, processed, question)
            except Exception as e:
                errors.append(f"Q&A failed: {e}")

        audio_url = None
        if generate_audio_summary and summary:
            try:
                audio_filename = f"{doc_id}_summary.mp3"
                audio.generate_summary_audio(
                    text=summary,
                    filename=audio_filename,
                    voice=audio_voice
                )
                audio_url = f"/audio/{audio_filename}"
            except Exception as e:
                errors.append(f"Audio failed: {e}")

        latency = round(time.time() - start, 2)

        return AnalyzeResponse(
            success=True,
            doc_id=doc_id,
            extracted_data=extracted_data,
            summary=summary,
            qa_result=qa_result,
            audio_summary_url=audio_url,
            metadata={
                "doc_id": doc_id,
                "filename": file.filename,
                "file_type": ext.lstrip("."),
                "pages_processed": processed.total_pages,
                "document_type": doc_type,
                "indexed": indexed,
                "latency_seconds": latency,
                "estimated_cost_usd": _estimate_total_cost(processed, extract_structured, generate_audio_summary)
            },
            errors=errors
        )

    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"Error processing {doc_id}: {e}", exc_info=True)
        raise HTTPException(500, f"Internal error: {str(e)}")
    finally:
        Path(file_path).unlink(missing_ok=True)

Endpoint POST /ask

@app.post("/ask")
async def ask_question(request: Request, body: QARequest):
    rag = request.app.state.rag

    if not body.question or len(body.question.strip()) < 5:
        raise HTTPException(400, "Question too short (minimum 5 characters)")

    try:
        result = rag.query(
            question=body.question,
            doc_id=body.doc_id
        )
        return result
    except Exception as e:
        logger.error(f"Error in Q&A: {e}", exc_info=True)
        raise HTTPException(500, f"Error generating the answer: {str(e)}")

Endpoint GET /health

@app.get("/health")
async def health_check(request: Request):
    from openai import OpenAI

    checks = {}

    try:
        client = OpenAI()
        client.models.list()
        checks["openai"] = "connected"
    except Exception as e:
        checks["openai"] = f"error: {str(e)}"

    try:
        rag = request.app.state.rag
        count = rag.collection.count()
        checks["chromadb"] = "connected"
        checks["indexed_documents"] = count
    except Exception as e:
        checks["chromadb"] = f"error: {str(e)}"

    uptime = round(time.time() - request.app.state.start_time, 1)
    checks["uptime_seconds"] = uptime

    all_ok = all(
        v == "connected" for k, v in checks.items()
        if k in ("openai", "chromadb")
    )

    status_code = 200 if all_ok else 503
    checks["status"] = "ok" if all_ok else "degraded"

    return JSONResponse(content=checks, status_code=status_code)

Endpoint GET /audio/{filename}

@app.get("/audio/{filename}")
async def serve_audio(filename: str):
    if not filename.endswith((".mp3", ".wav")):
        raise HTTPException(400, "Unsupported format")

    file_path = os.path.join("./audio_output", filename)
    if not os.path.exists(file_path):
        raise HTTPException(404, "Audio not found")

    return FileResponse(file_path, media_type="audio/mpeg", filename=filename)

Helper functions

def _generate_summary(analyzer, content) -> str:
    from openai import OpenAI
    client = OpenAI()

    if content.full_text:
        text = content.full_text[:6000]
    elif content.has_image_pages:
        descriptions = analyzer.describe_for_rag(content.get_images_for_vision()[:5])
        text = " ".join(descriptions)
    else:
        return "Document with no extractable content."

    r = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": f"Summarize this document in 3-5 key points:\n\n{text}"
        }],
        max_tokens=500
    )
    return r.choices[0].message.content


def _direct_qa(analyzer, content, question: str) -> dict:
    from openai import OpenAI
    client = OpenAI()

    if content.full_text:
        context = content.full_text[:6000]
    else:
        return {"question": question, "answer": "Document with no text for direct Q&A."}

    r = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": f"Context:\n{context}\n\nQuestion: {question}\n\nAnswer based only on the context."
        }],
        max_tokens=300
    )
    return {
        "question": question,
        "answer": r.choices[0].message.content,
        "sources": [],
        "confidence": None
    }


def _estimate_total_cost(content, extract: bool, audio: bool) -> float:
    cost = 0.002  # base: classification + summary
    if extract:
        cost += 0.01 if content.has_image_pages else 0.003
    if audio:
        cost += 0.01
    cost += content.image_page_count * 0.005  # image descriptions
    cost += 0.001  # embeddings
    return round(cost, 4)

Docker

Dockerfile

FROM python:3.11-slim AS base

WORKDIR /app

RUN apt-get update && \
    apt-get install -y --no-install-recommends \
        poppler-utils \
        ffmpeg \
    && rm -rf /var/lib/apt/lists/*

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

RUN mkdir -p /tmp/uploads /tmp/audio ./audio_output ./chroma_data

EXPOSE 8000

HEALTHCHECK --interval=30s --timeout=10s --retries=3 \
    CMD curl -f http://localhost:8000/health || exit 1

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "1"]

docker-compose.yml

version: "3.8"

services:
  document-analyzer:
    build: .
    ports:
      - "8000:8000"
    env_file:
      - .env
    volumes:
      - chroma_data:/app/chroma_data
      - audio_output:/app/audio_output
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 10s
      retries: 3

volumes:
  chroma_data:
  audio_output:

requirements.txt

openai>=1.0.0
pymupdf>=1.24.0
pillow>=10.0.0
pydantic>=2.0.0
chromadb>=0.5.0
fastapi>=0.110.0
uvicorn>=0.29.0
python-multipart>=0.0.9
python-dotenv>=1.0.0
pydub>=0.25.0
slowapi>=0.1.9

Build and run

docker build -t document-analyzer .

docker run -d \
  --name doc-analyzer \
  -p 8000:8000 \
  --env-file .env \
  -v doc_chroma:/app/chroma_data \
  -v doc_audio:/app/audio_output \
  document-analyzer

With docker-compose:

docker compose up -d
docker compose logs -f

Environment Variables

.env file

# Required
OPENAI_API_KEY=sk-...

# Optional — fallback providers
ANTHROPIC_API_KEY=sk-ant-...
GOOGLE_API_KEY=...

# Service configuration
LOG_LEVEL=INFO
MAX_FILE_SIZE_MB=50
REQUEST_TIMEOUT=120

# ChromaDB
CHROMA_PERSIST_DIR=./chroma_data

# Audio
AUDIO_OUTPUT_DIR=./audio_output
DEFAULT_TTS_VOICE=nova

Load configuration

from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    openai_api_key: str
    anthropic_api_key: str = ""
    google_api_key: str = ""
    log_level: str = "INFO"
    max_file_size_mb: int = 50
    request_timeout: int = 120
    chroma_persist_dir: str = "./chroma_data"
    audio_output_dir: str = "./audio_output"
    default_tts_voice: str = "nova"

    class Config:
        env_file = ".env"


settings = Settings()

Structured Logging

Configuration

import logging
import json
import sys


class JSONFormatter(logging.Formatter):
    def format(self, record):
        log_data = {
            "timestamp": self.formatTime(record),
            "level": record.levelname,
            "message": record.getMessage(),
            "module": record.module,
            "function": record.funcName,
        }
        if record.exc_info and record.exc_info[0]:
            log_data["exception"] = self.formatException(record.exc_info)
        return json.dumps(log_data)


def setup_logging(level: str = "INFO"):
    handler = logging.StreamHandler(sys.stdout)
    handler.setFormatter(JSONFormatter())

    root = logging.getLogger()
    root.setLevel(getattr(logging, level.upper()))
    root.addHandler(handler)

Logging middleware

@app.middleware("http")
async def log_requests(request: Request, call_next):
    start = time.time()
    request_id = str(uuid.uuid4())[:8]

    logger.info(json.dumps({
        "event": "request_start",
        "request_id": request_id,
        "method": request.method,
        "path": request.url.path,
        "client": request.client.host if request.client else "unknown"
    }))

    response = await call_next(request)

    duration = round(time.time() - start, 3)
    logger.info(json.dumps({
        "event": "request_end",
        "request_id": request_id,
        "status": response.status_code,
        "duration_seconds": duration
    }))

    response.headers["X-Request-ID"] = request_id
    response.headers["X-Duration-Seconds"] = str(duration)

    return response

Rate Limiting

from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)


@app.post("/analyze", response_model=AnalyzeResponse)
@limiter.limit("10/minute")
async def analyze_document(request: Request, ...):
    ...


@app.post("/ask")
@limiter.limit("30/minute")
async def ask_question(request: Request, ...):
    ...

Cost Optimization

Implemented strategies

Model selection per operation

MODEL_SELECTION = {
    "classify": "gpt-4o-mini",           # cheap, sufficient
    "extract_structured_text": "gpt-4o-mini",  # text → mini is enough
    "extract_structured_image": "gpt-4o",      # vision → needs 4o
    "summarize": "gpt-4o-mini",           # summary → mini is enough
    "describe_image": "gpt-4o-mini",      # description → mini is enough
    "qa_generate": "gpt-4o",              # detailed answer → 4o is better
    "embeddings": "text-embedding-3-small",  # cheapest
}

Simple cache

import hashlib
from functools import lru_cache


class ResultCache:
    def __init__(self):
        self._cache: dict[str, any] = {}

    def get(self, key: str):
        return self._cache.get(key)

    def set(self, key: str, value):
        self._cache[key] = value

    def make_key(self, *args) -> str:
        content = "|".join(str(a)[:200] for a in args)
        return hashlib.md5(content.encode()).hexdigest()


cache = ResultCache()


def classify_with_cache(analyzer, content) -> str:
    cache_key = cache.make_key("classify", content.file_path, content.total_pages)
    cached = cache.get(cache_key)
    if cached:
        return cached

    result = analyzer.classify(content)
    cache.set(cache_key, result)
    return result

Cost Monitoring

Cost tracker

class CostTracker:
    def __init__(self):
        self.operations: list[dict] = []

    def track(self, operation: str, model: str, tokens_in: int = 0, tokens_out: int = 0):
        cost = self._calculate_cost(model, tokens_in, tokens_out)
        self.operations.append({
            "operation": operation,
            "model": model,
            "tokens_in": tokens_in,
            "tokens_out": tokens_out,
            "cost_usd": cost,
            "timestamp": time.time()
        })

    def get_total(self) -> float:
        return sum(op["cost_usd"] for op in self.operations)

    def get_breakdown(self) -> dict:
        by_operation = {}
        for op in self.operations:
            name = op["operation"]
            if name not in by_operation:
                by_operation[name] = {"count": 0, "cost": 0}
            by_operation[name]["count"] += 1
            by_operation[name]["cost"] += op["cost_usd"]
        return by_operation

    def _calculate_cost(self, model: str, tokens_in: int, tokens_out: int) -> float:
        rates = {
            "gpt-4o": {"in": 2.50 / 1_000_000, "out": 10.00 / 1_000_000},
            "gpt-4o-mini": {"in": 0.15 / 1_000_000, "out": 0.60 / 1_000_000},
            "text-embedding-3-small": {"in": 0.02 / 1_000_000, "out": 0},
        }
        rate = rates.get(model, {"in": 0.001, "out": 0.002})
        return round(tokens_in * rate["in"] + tokens_out * rate["out"], 6)

Troubleshooting

"Docker build fails with a PyMuPDF error"

Cause: PyMuPDF needs to be compiled and the slim image doesn't have build tools.

Solution: Use the image with build tools or install dependencies:

RUN apt-get update && apt-get install -y --no-install-recommends \
    build-essential \
    && pip install --no-cache-dir -r requirements.txt \
    && apt-get remove -y build-essential \
    && apt-get autoremove -y \
    && rm -rf /var/lib/apt/lists/*

"The service runs out of memory with large PDFs"

Solution: Limit uvicorn workers and add swap:

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "1", "--limit-max-requests", "100"]

"OpenAI rate limit (429) in production"

Solution: Implement retry with exponential backoff:

import time


def retry_with_backoff(fn, max_retries: int = 3, base_delay: float = 1.0):
    for attempt in range(max_retries):
        try:
            return fn()
        except Exception as e:
            if "429" in str(e) and attempt < max_retries - 1:
                delay = base_delay * (2 ** attempt)
                logger.warning(f"Rate limit, retry in {delay}s (attempt {attempt + 1})")
                time.sleep(delay)
            else:
                raise

"ChromaDB loses data when the container restarts"

Solution: Mount a volume for persistence:

volumes:
  - chroma_data:/app/chroma_data

Local Execution

Without Docker

source venv/bin/activate
uvicorn main:app --reload --host 0.0.0.0 --port 8000

Test with curl

curl -X POST "http://localhost:8000/analyze" \
  -F "file=@invoice_example.pdf" \
  -F "question=What is the total?" \
  -F "extract_structured=true" \
  -F "generate_summary=true" \
  -F "generate_audio_summary=true" \
  -F "audio_voice=nova"

curl http://localhost:8000/health

curl -X POST "http://localhost:8000/ask" \
  -H "Content-Type: application/json" \
  -d '{"question": "Who is the vendor?", "doc_id": "doc-id-here"}'

Test with Python

import requests

with open("invoice_example.pdf", "rb") as f:
    response = requests.post(
        "http://localhost:8000/analyze",
        files={"file": ("invoice.pdf", f, "application/pdf")},
        data={
            "question": "What is the total?",
            "extract_structured": "true",
            "generate_summary": "true",
            "generate_audio_summary": "false"
        }
    )

result = response.json()
print(f"Success: {result['success']}")
print(f"Type: {result['metadata']['document_type']}")
print(f"Summary: {result['summary']}")
if result.get("qa_result"):
    print(f"Answer: {result['qa_result']['answer']}")

Exercises

Exercise 1: Rate limiting per API key

Implement differentiated rate limiting: users with a premium API key get 50 requests/minute, users without a key get 5 requests/minute. Use slowapi with a custom key function.

from slowapi import Limiter


def get_rate_limit_key(request: Request) -> str:
    api_key = request.headers.get("X-API-Key", "")
    if api_key:
        return f"premium_{api_key}"
    return get_remote_address(request)


def get_rate_limit(request: Request) -> str:
    api_key = request.headers.get("X-API-Key", "")
    if api_key and api_key.startswith("premium_"):
        return "50/minute"
    return "5/minute"


limiter = Limiter(key_func=get_rate_limit_key)
app.state.limiter = limiter


@app.post("/analyze")
@limiter.limit(lambda: "50/minute", key_func=lambda request: (
    f"premium_{request.headers.get('X-API-Key', '')}"
    if request.headers.get("X-API-Key")
    else get_remote_address(request)
))
async def analyze_document(request: Request, ...):
    ...

Exercise 2: Cost dashboard

Implement a GET /costs endpoint that returns: total accumulated cost, cost per operation, cost per model, and the last 10 requests with their individual cost.

cost_tracker = CostTracker()


@app.get("/costs")
async def get_costs():
    breakdown = cost_tracker.get_breakdown()

    by_model = {}
    for op in cost_tracker.operations:
        model = op["model"]
        if model not in by_model:
            by_model[model] = {"count": 0, "cost": 0}
        by_model[model]["count"] += 1
        by_model[model]["cost"] += op["cost_usd"]

    recent = sorted(cost_tracker.operations, key=lambda x: x["timestamp"], reverse=True)[:10]

    return {
        "total_cost_usd": round(cost_tracker.get_total(), 4),
        "total_operations": len(cost_tracker.operations),
        "by_operation": {
            k: {"count": v["count"], "cost_usd": round(v["cost"], 4)}
            for k, v in breakdown.items()
        },
        "by_model": {
            k: {"count": v["count"], "cost_usd": round(v["cost"], 4)}
            for k, v in by_model.items()
        },
        "recent_operations": [
            {
                "operation": op["operation"],
                "model": op["model"],
                "cost_usd": round(op["cost_usd"], 6),
                "tokens": op["tokens_in"] + op["tokens_out"]
            }
            for op in recent
        ]
    }

Summary

• FastAPI exposes 4 endpoints: /analyze, /ask, /health, /audio/{filename}.
• Docker packages the service with all dependencies (poppler, ffmpeg).
• Environment variables manage API keys and configuration without hardcoding.
• Structured logging in JSON makes debugging and monitoring easier.
• Rate limiting protects against abuse (slowapi).
• Cost optimization: model selection per operation, cache, resolution reduction.
• Cost tracker records each operation for monitoring and budgeting.
• The system is resilient: partial errors don't crash the whole request.

Additional Resources

1. FastAPI Deployment — Official guide
2. Docker Best Practices — Optimized Dockerfile
3. slowapi — Rate limiting for FastAPI
4. OpenAI Rate Limits — Limits per tier
5. Pydantic Settings — Typed configuration