2. Technical Specifications

Description

Before writing a single line of code, you need a clear contract: what comes in, what goes out, which components participate, how they communicate, and how much it costs. This capsule defines the complete technical specifications of the Multimodal Document Analyzer. It's the equivalent of a technical design document (TDD) that, on an engineering team, is reviewed before implementing.

Why it matters: Building without specifications leads to ad-hoc decisions, constant refactors, and components that don't fit together. Defining inputs, outputs, data models and endpoints before implementing lets you validate the design, estimate costs, and ensure all modules speak the same language.

Connection with the module: This capsule is the reference for capsules 03-08. Every component you build must respect the contracts defined here. If you need to change a specification during implementation, come back here, update the contract, and propagate the change.

System Inputs

1. Document (required)

The main input is a document file. The system supports two categories:

Technical constraints

SUPPORTED_EXTENSIONS = {".pdf", ".png", ".jpg", ".jpeg", ".webp"}
MAX_FILE_SIZE_BYTES = 50 * 1024 * 1024  # 50 MB
MAX_PDF_PAGES = 50
MAX_IMAGE_DIMENSION = 4096  # pixels per side (recommended)

Content types inside PDFs

A PDF can have different types of pages:

Per-page type detection is critical: a 20-page PDF can have 15 with text and 5 scanned. The DocumentProcessor must handle both cases within the same document.

2. Question (optional)

For the Q&A module:

QUESTION_MAX_LENGTH = 500  # characters
QUESTION_MIN_LENGTH = 5    # minimum characters for a valid question

If no question is sent, the system processes the document without Q&A. If one is sent, the document is automatically indexed to answer it.

3. Processing options

Flags that control which operations to run:

Pydantic model of the request

from pydantic import BaseModel, Field
from typing import Optional
from enum import Enum


class TTSVoice(str, Enum):
    ALLOY = "alloy"
    ECHO = "echo"
    FABLE = "fable"
    ONYX = "onyx"
    NOVA = "nova"
    SHIMMER = "shimmer"


class AnalyzeRequest(BaseModel):
    question: Optional[str] = Field(
        None, min_length=5, max_length=500,
        description="Question about the document for Q&A"
    )
    extract_structured: bool = Field(
        True, description="Extract structured data from the document"
    )
    generate_summary: bool = Field(
        True, description="Generate an executive summary"
    )
    generate_audio_summary: bool = Field(
        False, description="Generate an audio summary (TTS)"
    )
    index_for_qa: bool = Field(
        True, description="Index for later Q&A"
    )
    audio_voice: TTSVoice = Field(
        TTSVoice.NOVA, description="Voice for audio synthesis"
    )

System Outputs

Standard response

Every response from the /analyze endpoint follows this format:

class DocumentMetadata(BaseModel):
    doc_id: str
    filename: str
    file_type: str
    pages_processed: int
    document_type: Optional[str] = None
    indexed: bool = False
    latency_seconds: float
    estimated_cost_usd: float


class StructuredData(BaseModel):
    document_type: str
    fields: dict
    confidence: Optional[float] = None


class QAResult(BaseModel):
    question: str
    answer: str
    sources: list[str] = []
    confidence: Optional[float] = None


class AnalyzeResponse(BaseModel):
    success: bool
    doc_id: str
    extracted_data: Optional[StructuredData] = None
    summary: Optional[str] = None
    qa_result: Optional[QAResult] = None
    audio_summary_url: Optional[str] = None
    metadata: DocumentMetadata
    errors: list[str] = []

Example of a complete response

{
    "success": true,
    "doc_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
    "extracted_data": {
        "document_type": "invoice",
        "fields": {
            "date": "2025-03-15",
            "invoice_number": "FAC-2025-0042",
            "vendor": "Tech Solutions S.A.",
            "subtotal": 1500.00,
            "tax": 240.00,
            "total": 1740.00,
            "items": [
                {"description": "Software license", "quantity": 1, "price": 1200.00},
                {"description": "Technical support", "quantity": 1, "price": 300.00}
            ]
        },
        "confidence": 0.95
    },
    "summary": "Invoice FAC-2025-0042 from Tech Solutions S.A. for $1,740.00 MXN. Includes software license ($1,200) and technical support ($300). Subtotal $1,500 + tax $240. Date: March 15, 2025.",
    "qa_result": {
        "question": "What is the invoice total?",
        "answer": "The invoice total is $1,740.00 MXN, which includes a subtotal of $1,500.00 plus tax of $240.00.",
        "sources": ["Page 1: Total: $1,740.00", "Page 1: Subtotal: $1,500.00 | Tax: $240.00"],
        "confidence": 0.98
    },
    "audio_summary_url": "/audio/a1b2c3d4_summary.mp3",
    "metadata": {
        "doc_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
        "filename": "invoice_march.pdf",
        "file_type": "pdf",
        "pages_processed": 1,
        "document_type": "invoice",
        "indexed": true,
        "latency_seconds": 8.45,
        "estimated_cost_usd": 0.045
    },
    "errors": []
}

Response with partial errors

The system doesn't fail completely if one component has problems. It reports partial success:

{
    "success": true,
    "doc_id": "...",
    "extracted_data": null,
    "summary": "Document summary...",
    "qa_result": null,
    "audio_summary_url": null,
    "metadata": {
        "doc_id": "...",
        "filename": "technical_manual.pdf",
        "file_type": "pdf",
        "pages_processed": 15,
        "document_type": "manual",
        "indexed": true,
        "latency_seconds": 22.1,
        "estimated_cost_usd": 0.08
    },
    "errors": [
        "Structured extraction not available for type 'manual'",
        "TTS failed: summary exceeds character limit"
    ]
}

System Components

Component table

Interface of each component

Each component exposes a minimal, predictable interface:

class DocumentProcessor:
    def process(self, file_path: str) -> ProcessedDocument: ...

class VisionAnalyzer:
    def classify(self, content: ProcessedDocument) -> str: ...
    def extract_structured(self, content: ProcessedDocument, doc_type: str) -> StructuredData: ...

class RAGModule:
    def index(self, doc_id: str, content: ProcessedDocument) -> None: ...
    def query(self, question: str, doc_id: str = None) -> QAResult: ...

class AudioModule:
    def generate_summary_audio(self, text: str, voice: str = "nova") -> str: ...

class DocumentSummarizer:
    def summarize(self, content: ProcessedDocument) -> str: ...

class CostTracker:
    def track(self, operation: str, model: str, tokens: int) -> None: ...
    def get_total(self) -> float: ...

Intermediate data model

The components communicate through a shared model:

class PageContent(BaseModel):
    page_number: int
    content_type: str  # "text" | "image"
    text: Optional[str] = None
    image_base64: Optional[str] = None


class ProcessedDocument(BaseModel):
    file_path: str
    file_type: str  # "pdf" | "image"
    total_pages: int
    pages: list[PageContent]
    full_text: Optional[str] = None
    has_text_pages: bool = False
    has_image_pages: bool = False


class DocumentChunk(BaseModel):
    chunk_id: str
    doc_id: str
    text: str
    page_number: int
    content_type: str  # "text" | "image_description"
    metadata: dict = {}

REST API (FastAPI)

Endpoints

POST /analyze

Main endpoint. Receives a document and options, returns the complete analysis.

POST /analyze
Content-Type: multipart/form-data

Fields:
  - file: UploadFile (PDF or image, required)
  - question: str (optional, max 500 chars)
  - extract_structured: bool (default: true)
  - generate_summary: bool (default: true)
  - generate_audio_summary: bool (default: false)
  - index_for_qa: bool (default: true)
  - audio_voice: str (default: "nova")

Response: 200 OK → AnalyzeResponse
Errors:
  - 400: Invalid file (format, size)
  - 413: File exceeds maximum size
  - 422: Incorrect validation parameters
  - 500: Internal processing error
  - 503: Service unavailable (invalid API keys)

POST /ask

To ask questions about already-indexed documents:

POST /ask
Content-Type: application/json

Body:
{
    "question": "What is the due date?",
    "doc_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890"  // optional
}

Response: 200 OK → QAResult
Errors:
  - 400: Empty or invalid question
  - 404: doc_id not found (if specified)
  - 500: Error generating the answer

GET /health

GET /health

Response: 200 OK
{
    "status": "ok",
    "openai": "connected",
    "chromadb": "connected",
    "indexed_documents": 5,
    "uptime_seconds": 3600
}

Degraded response: 503
{
    "status": "degraded",
    "openai": "error: invalid API key",
    "chromadb": "connected"
}

Architecture Diagram

Full flow of a POST /analyze request

Client
  │
  ├─── POST /analyze (file=invoice.pdf, question="Total?", extract=true, audio=true)
  │
  ▼
┌─────────────────────────────────────────────────────────────┐
│                    FastAPI Router                            │
│  1. Validate file (type, size)                             │
│  2. Save temporary file                                     │
│  3. Generate doc_id                                        │
│  4. Start CostTracker                                      │
└──────────┬──────────────────────────────────────────────────┘
           │
           ▼
┌─────────────────────────────────────────────────────────────┐
│              DocumentProcessor.process()                     │
│  - Detect type (PDF vs image)                              │
│  - For PDF: iterate pages                                   │
│    - Page with text → extract text                          │
│    - Scanned page → convert to base64 image                 │
│  - For image: encode to base64                             │
│  - Return ProcessedDocument                                 │
└──────────┬──────────────────────────────────────────────────┘
           │
     ┌─────┴─────────────────────────────────────┐
     │                                           │
     ▼                                           ▼
┌──────────────────────┐              ┌──────────────────────┐
│ VisionAnalyzer       │              │ RAGModule            │
│ .classify() → type   │              │ .index() → ChromaDB  │
│ .extract_structured()│              │                      │
│  → StructuredData    │              │ If there's a question:│
└──────────┬───────────┘              │ .query() → QAResult  │
           │                          └──────────┬───────────┘
           ▼                                     │
┌──────────────────────┐                         │
│ DocumentSummarizer   │                         │
│ .summarize()         │                         │
│  → text summary      │                         │
└──────────┬───────────┘                         │
           │                                     │
           ▼                                     │
┌──────────────────────┐                         │
│ AudioModule          │                         │
│ .generate_summary_   │                         │
│  audio()             │                         │
│  → .mp3 file         │                         │
└──────────┬───────────┘                         │
           │                                     │
           └─────────────┬───────────────────────┘
                         │
                         ▼
              ┌──────────────────────┐
              │  Build               │
              │  AnalyzeResponse     │
              │  + metadata          │
              │  + costs             │
              └──────────┬───────────┘
                         │
                         ▼
                     200 OK → JSON

Technology Stack

Main dependencies

Optional dependencies

AI models used

Cost Budget

Per operation

Cost scenarios

Recommended monthly limit

For a production service with ~1000 documents/month:

1000 docs × $0.05 average = $50/month
+ 20% buffer for retries   = $60/month

Extraction Schemas by Document Type

Invoice

INVOICE_SCHEMA = {
    "date": "str (YYYY-MM-DD)",
    "invoice_number": "str",
    "vendor": "str",
    "recipient": "str",
    "subtotal": "float",
    "tax": "float",
    "total": "float",
    "currency": "str (MXN, USD, EUR)",
    "items": [
        {
            "description": "str",
            "quantity": "int",
            "unit_price": "float",
            "line_total": "float"
        }
    ]
}

Contract

CONTRACT_SCHEMA = {
    "contract_type": "str",
    "parties": ["str"],
    "signing_date": "str (YYYY-MM-DD)",
    "effective_date": "str (YYYY-MM-DD)",
    "subject": "str",
    "amount": "float or null",
    "key_clauses": ["str"]
}

Manual / Report

REPORT_SCHEMA = {
    "title": "str",
    "author": "str or null",
    "date": "str (YYYY-MM-DD) or null",
    "sections": ["str"],
    "executive_summary": "str"
}

Schema registry

EXTRACTION_SCHEMAS: dict[str, dict] = {
    "invoice": INVOICE_SCHEMA,
    "contract": CONTRACT_SCHEMA,
    "manual": REPORT_SCHEMA,
    "report": REPORT_SCHEMA,
}

def get_schema_for_type(doc_type: str) -> dict:
    return EXTRACTION_SCHEMAS.get(
        doc_type.lower(),
        {"general_content": "str", "key_points": ["str"]}
    )

Exercises

Exercise 1: Complete input validator

Implement a function that validates the input file by checking: existence, supported extension, maximum size, and for PDFs, maximum number of pages. It must return a list of errors (empty if everything is valid).

import fitz
from pathlib import Path


def validate_input(file_path: str) -> list[str]:
    errors = []
    p = Path(file_path)

    if not p.exists():
        return ["File not found"]

    if p.suffix.lower() not in SUPPORTED_EXTENSIONS:
        errors.append(
            f"Format '{p.suffix}' not supported. "
            f"Valid formats: {', '.join(SUPPORTED_EXTENSIONS)}"
        )
        return errors

    file_size = p.stat().st_size
    if file_size > MAX_FILE_SIZE_BYTES:
        errors.append(
            f"File of {file_size / 1024 / 1024:.1f} MB exceeds "
            f"the limit of {MAX_FILE_SIZE_BYTES / 1024 / 1024:.0f} MB"
        )

    if file_size == 0:
        errors.append("Empty file")
        return errors

    if p.suffix.lower() == ".pdf":
        try:
            doc = fitz.open(file_path)
            if len(doc) > MAX_PDF_PAGES:
                errors.append(
                    f"PDF of {len(doc)} pages exceeds "
                    f"the limit of {MAX_PDF_PAGES} pages"
                )
            if len(doc) == 0:
                errors.append("PDF with no pages")
            doc.close()
        except Exception as e:
            errors.append(f"Corrupt or unreadable PDF: {e}")

    return errors

Exercise 2: Complete Pydantic model for the response

Define all the Pydantic models needed for the /analyze endpoint response, including custom validations: doc_id must be a valid UUID, latency_seconds must be positive, estimated_cost_usd non-negative.

from pydantic import BaseModel, Field, field_validator
from typing import Optional
import uuid


class DocumentMetadata(BaseModel):
    doc_id: str
    filename: str
    file_type: str
    pages_processed: int = Field(ge=0)
    document_type: Optional[str] = None
    indexed: bool = False
    latency_seconds: float = Field(ge=0)
    estimated_cost_usd: float = Field(ge=0)

    @field_validator("doc_id")
    @classmethod
    def validate_uuid(cls, v: str) -> str:
        uuid.UUID(v)
        return v

    @field_validator("file_type")
    @classmethod
    def validate_file_type(cls, v: str) -> str:
        if v not in ("pdf", "png", "jpg", "jpeg", "webp"):
            raise ValueError(f"Invalid file type: {v}")
        return v


class StructuredData(BaseModel):
    document_type: str
    fields: dict
    confidence: Optional[float] = Field(None, ge=0, le=1)


class QAResult(BaseModel):
    question: str = Field(min_length=1)
    answer: str
    sources: list[str] = []
    confidence: Optional[float] = Field(None, ge=0, le=1)


class AnalyzeResponse(BaseModel):
    success: bool
    doc_id: str
    extracted_data: Optional[StructuredData] = None
    summary: Optional[str] = None
    qa_result: Optional[QAResult] = None
    audio_summary_url: Optional[str] = None
    metadata: DocumentMetadata
    errors: list[str] = []

Summary

• The Document Analyzer accepts PDFs (up to 50 pages) and images (up to 20 MB) as input.
• The response includes: structured data, summary, Q&A with sources, audio URL, and metadata with costs.
• 6 components with defined interfaces: DocumentProcessor, VisionAnalyzer, RAGModule, AudioModule, Summarizer, CostTracker.
• 5 REST endpoints: /analyze (main), /ask (Q&A), /documents/{id}, /health, /audio/{filename}.
• Extraction schemas vary by document type: invoice, contract, manual/report.
• Average cost: $0.03-0.08 per document. ~$50-60/month for 1000 documents.
• All Pydantic models are defined for strict validation of inputs and outputs.

Additional Resources

1. FastAPI Request Files — File upload
2. Pydantic V2 — Model validation
3. OpenAI Pricing — Up-to-date model costs
4. PyMuPDF — PDF processing