8. Project: Multimodal Document Analyzer

Description

This is the final project of the Multimodal AI Guide. It integrates all the components you built in the previous capsules — DocumentProcessor, VisionAnalyzer, RAGModule, AudioModule — into a single DocumentAnalyzer class that orchestrates the complete pipeline. It includes: the integrated, working code, a demo script you can run, test cases to validate each component, a deployment checklist, and the final evaluation.

What you deliver: A complete system that receives a PDF or image, processes it, extracts structured data, generates a summary, indexes for Q&A, answers questions, and optionally generates audio — all through a REST API or directly as a Python module.

Success criterion: If you can run the demo script and get correct results for a sample PDF, your project is complete. If you can also run docker build and run curl against the API, you're production-ready.

Final Architecture

DocumentAnalyzer class: the orchestrator

DocumentAnalyzer
├── analyze()          → Full pipeline: process + classify + extract + summarize + index + Q&A + audio
├── ask()              → Q&A over already-indexed documents
├── get_document()     → Retrieve the result of a previous analysis
├── list_documents()   → List processed documents
└── health_check()     → Check the state of services

Internal flow of analyze()

analyze(file_path, question, options)
    │
    ├── 1. Validate input
    │      └── Extension, size, pages
    │
    ├── 2. DocumentProcessor.process()
    │      └── Text/images per page
    │
    ├── 3. VisionAnalyzer.classify()
    │      └── invoice | contract | manual | report | other
    │
    ├── 4. VisionAnalyzer.extract_structured() [if extract_structured=True]
    │      └── Typed data based on classification
    │
    ├── 5. Summarizer [if generate_summary=True]
    │      └── Executive summary in 3-5 points
    │
    ├── 6. RAGModule.index() [if index_for_qa=True]
    │      └── Text + image descriptions → ChromaDB
    │
    ├── 7. RAGModule.query() [if there's a question]
    │      └── Answer with cited sources
    │
    ├── 8. AudioModule.generate_summary_audio() [if generate_audio=True]
    │      └── .mp3 file
    │
    └── 9. Build AnalyzeResponse with metadata and costs

Complete Implementation: DocumentAnalyzer

Main class

import json
import logging
import os
import time
import uuid
from pathlib import Path
from typing import Optional

from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

logger = logging.getLogger(__name__)


SUPPORTED_EXTENSIONS = {".pdf", ".png", ".jpg", ".jpeg", ".webp"}
MAX_FILE_SIZE_BYTES = 50 * 1024 * 1024
MAX_PDF_PAGES = 50
TEXT_THRESHOLD = 50


class DocumentAnalyzer:
    def __init__(
        self,
        chroma_persist_dir: Optional[str] = "./chroma_data",
        audio_output_dir: str = "./audio_output"
    ):
        from modules.document_processor import DocumentProcessor
        from modules.vision_analyzer import VisionAnalyzer
        from modules.rag_module import RAGModule
        from modules.audio_module import AudioModule

        self.processor = DocumentProcessor()
        self.analyzer = VisionAnalyzer()
        self.rag = RAGModule(persist_directory=chroma_persist_dir)
        self.audio = AudioModule(output_dir=audio_output_dir)
        self.openai_client = OpenAI()
        self.results_cache: dict[str, dict] = {}

        logger.info("DocumentAnalyzer initialized")

    def analyze(
        self,
        file_path: str,
        question: Optional[str] = None,
        extract_structured: bool = True,
        generate_summary: bool = True,
        generate_audio_summary: bool = False,
        index_for_qa: bool = True,
        audio_voice: str = "nova"
    ) -> dict:
        start = time.time()
        doc_id = str(uuid.uuid4())
        errors: list[str] = []

        validation_errors = self._validate_input(file_path)
        if validation_errors:
            return {
                "success": False,
                "doc_id": doc_id,
                "errors": validation_errors,
                "metadata": {"latency_seconds": round(time.time() - start, 2)}
            }

        try:
            content = self.processor.process(file_path)
        except Exception as e:
            return {
                "success": False,
                "doc_id": doc_id,
                "errors": [f"Error processing document: {e}"],
                "metadata": {"latency_seconds": round(time.time() - start, 2)}
            }

        doc_type = "other"
        try:
            doc_type = self.analyzer.classify(content)
            logger.info(f"[{doc_id}] Classified as: {doc_type}")
        except Exception as e:
            errors.append(f"Classification failed: {e}")

        extracted_data = None
        if extract_structured:
            try:
                extraction = self.analyzer.extract_structured(content, doc_type)
                extracted_data = {
                    "document_type": extraction.document_type,
                    "fields": extraction.fields,
                    "confidence": extraction.confidence
                }
                logger.info(f"[{doc_id}] Extraction: confidence {extraction.confidence}")
            except Exception as e:
                errors.append(f"Extraction failed: {e}")

        summary = None
        if generate_summary:
            try:
                summary = self._generate_summary(content)
                logger.info(f"[{doc_id}] Summary generated: {len(summary)} chars")
            except Exception as e:
                errors.append(f"Summary failed: {e}")

        indexed = False
        if index_for_qa:
            try:
                image_descriptions = None
                if content.has_image_pages:
                    image_descriptions = self.analyzer.describe_for_rag(
                        content.get_images_for_vision()
                    )

                self.rag.index(
                    doc_id=doc_id,
                    content=content,
                    filename=Path(file_path).name,
                    document_type=doc_type,
                    image_descriptions=image_descriptions
                )
                indexed = True
                logger.info(f"[{doc_id}] Indexed in RAG")
            except Exception as e:
                errors.append(f"Indexing failed: {e}")

        qa_result = None
        if question:
            try:
                if indexed:
                    qa = self.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 = self._direct_qa(content, question)
                logger.info(f"[{doc_id}] Q&A completed")
            except Exception as e:
                errors.append(f"Q&A failed: {e}")

        audio_path = None
        if generate_audio_summary and summary:
            try:
                audio_filename = f"{doc_id}_summary.mp3"
                audio_path = self.audio.generate_summary_audio(
                    text=summary,
                    filename=audio_filename,
                    voice=audio_voice
                )
                logger.info(f"[{doc_id}] Audio generated: {audio_path}")
            except Exception as e:
                errors.append(f"Audio failed: {e}")

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

        result = {
            "success": True,
            "doc_id": doc_id,
            "extracted_data": extracted_data,
            "summary": summary,
            "qa_result": qa_result,
            "audio_summary_path": audio_path,
            "metadata": {
                "doc_id": doc_id,
                "filename": Path(file_path).name,
                "file_type": Path(file_path).suffix.lstrip("."),
                "pages_processed": content.total_pages,
                "text_pages": content.text_page_count,
                "image_pages": content.image_page_count,
                "document_type": doc_type,
                "indexed": indexed,
                "latency_seconds": latency,
                "estimated_cost_usd": self._estimate_cost(
                    content, extract_structured, generate_audio_summary
                )
            },
            "errors": errors
        }

        self.results_cache[doc_id] = result
        return result

    def ask(self, question: str, doc_id: Optional[str] = None) -> dict:
        qa = self.rag.query(question=question, doc_id=doc_id)
        return {
            "question": qa.question,
            "answer": qa.answer,
            "sources": qa.sources,
            "confidence": qa.confidence
        }

    def get_document(self, doc_id: str) -> Optional[dict]:
        return self.results_cache.get(doc_id)

    def list_documents(self) -> list[dict]:
        return self.rag.list_documents()

    def health_check(self) -> dict:
        checks = {}
        try:
            self.openai_client.models.list()
            checks["openai"] = "connected"
        except Exception as e:
            checks["openai"] = f"error: {e}"

        try:
            checks["chromadb"] = "connected"
            checks["indexed_chunks"] = self.rag.collection.count()
        except Exception as e:
            checks["chromadb"] = f"error: {e}"

        checks["status"] = (
            "ok" if checks.get("openai") == "connected" else "degraded"
        )
        return checks

    def _validate_input(self, 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")
            return errors

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

        if p.stat().st_size == 0:
            errors.append("Empty file")

        return errors

    def _generate_summary(self, content) -> str:
        if content.full_text:
            text = content.full_text[:6000]
        elif content.has_image_pages:
            descriptions = self.analyzer.describe_for_rag(
                content.get_images_for_vision()[:5]
            )
            text = " ".join(descriptions)
        else:
            return "Document with no extractable content."

        r = self.openai_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(self, content, question: str) -> dict:
        if not content.full_text:
            return {
                "question": question,
                "answer": "No text available to answer.",
                "sources": [],
                "confidence": 0
            }

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

    def _estimate_cost(self, content, extract: bool, audio: bool) -> float:
        cost = 0.002  # 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
        cost += 0.001  # embeddings
        return round(cost, 4)

Individual Modules (Summary)

For the DocumentAnalyzer to work, you need the modules in modules/. Here is the structure and a summary of each one. The complete code is in capsules 03-06.

modules/document_processor.py

import base64
import logging
from pathlib import Path
from typing import Optional

import fitz
from pydantic import BaseModel

logger = logging.getLogger(__name__)


class PageContent(BaseModel):
    page_number: int
    content_type: str
    text: Optional[str] = None
    image_base64: Optional[str] = None
    char_count: int = 0
    image_size_bytes: int = 0


class ProcessedDocument(BaseModel):
    file_path: str
    file_type: str
    total_pages: int
    pages: list[PageContent]
    full_text: Optional[str] = None
    has_text_pages: bool = False
    has_image_pages: bool = False
    text_page_count: int = 0
    image_page_count: int = 0

    def get_text_pages(self) -> list[PageContent]:
        return [p for p in self.pages if p.content_type == "text"]

    def get_image_pages(self) -> list[PageContent]:
        return [p for p in self.pages if p.content_type == "image"]

    def get_images_for_vision(self) -> list[dict]:
        return [
            {"page": p.page_number, "base64": p.image_base64}
            for p in self.get_image_pages() if p.image_base64
        ]


class DocumentProcessor:
    def __init__(self, text_threshold: int = 50, dpi_scale: float = 150/72):
        self.text_threshold = text_threshold
        self.dpi_scale = dpi_scale

    def process(self, file_path: str) -> ProcessedDocument:
        path = Path(file_path)
        if path.suffix.lower() == ".pdf":
            return self._process_pdf(file_path)
        return self._process_image(file_path)

    def _process_pdf(self, file_path: str) -> ProcessedDocument:
        doc = fitz.open(file_path)
        pages = []
        text_parts = []
        text_count = 0
        image_count = 0

        try:
            for i in range(len(doc)):
                page = doc[i]
                text = page.get_text()

                if len(text.strip()) > self.text_threshold:
                    pages.append(PageContent(
                        page_number=i + 1, content_type="text",
                        text=text, char_count=len(text)
                    ))
                    text_parts.append(text)
                    text_count += 1
                else:
                    mat = fitz.Matrix(self.dpi_scale, self.dpi_scale)
                    pix = page.get_pixmap(matrix=mat, alpha=False)
                    b64 = base64.b64encode(pix.tobytes("png")).decode()
                    pages.append(PageContent(
                        page_number=i + 1, content_type="image",
                        image_base64=b64, image_size_bytes=len(b64) * 3 // 4
                    ))
                    image_count += 1
        finally:
            total = len(doc)
            doc.close()

        return ProcessedDocument(
            file_path=file_path, file_type="pdf", total_pages=total,
            pages=pages,
            full_text="\n\n".join(text_parts) if text_parts else None,
            has_text_pages=text_count > 0, has_image_pages=image_count > 0,
            text_page_count=text_count, image_page_count=image_count
        )

    def _process_image(self, file_path: str) -> ProcessedDocument:
        with open(file_path, "rb") as f:
            raw = f.read()
        b64 = base64.b64encode(raw).decode()

        return ProcessedDocument(
            file_path=file_path, file_type="image", total_pages=1,
            pages=[PageContent(
                page_number=1, content_type="image",
                image_base64=b64, image_size_bytes=len(raw)
            )],
            has_text_pages=False, has_image_pages=True,
            text_page_count=0, image_page_count=1
        )

modules/vision_analyzer.py (interface)

import json
import logging
import os
from typing import Optional

from openai import OpenAI
from pydantic import BaseModel, Field

logger = logging.getLogger(__name__)


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


SCHEMA_PROMPTS = {
    "invoice": "date, invoice_number, vendor, recipient, subtotal, tax, total, currency, items",
    "contract": "contract_type, parties, signing_date, effective_date, subject, amount, key_clauses",
    "manual": "title, author, date, sections, executive_summary",
    "report": "title, author, date, sections, executive_summary",
}


class VisionAnalyzer:
    def __init__(self):
        self.client = OpenAI()
        self.anthropic_client = None
        self._init_fallbacks()

    def _init_fallbacks(self):
        try:
            import anthropic
            if os.getenv("ANTHROPIC_API_KEY"):
                self.anthropic_client = anthropic.Anthropic()
        except ImportError:
            pass

    def classify(self, content) -> str:
        if content.has_image_pages:
            images = content.get_images_for_vision()
            return self._classify_image(images[0]["base64"])
        if content.full_text:
            return self._classify_text(content.full_text[:2000])
        return "other"

    def _classify_image(self, image_b64: str) -> str:
        r = self.client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": [
                {"type": "text", "text": "Classify: invoice, contract, manual, report, other. Just the name."},
                {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{image_b64}"}}
            ]}],
            max_tokens=20, temperature=0
        )
        return r.choices[0].message.content.strip().lower()

    def _classify_text(self, text: str) -> str:
        r = self.client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": f"Classify: invoice, contract, manual, report, other. Just the name.\n\n{text}"}],
            max_tokens=20, temperature=0
        )
        return r.choices[0].message.content.strip().lower()

    def extract_structured(self, content, doc_type: str) -> ExtractionResult:
        schema = SCHEMA_PROMPTS.get(doc_type, "main_content, key_points")

        if content.has_image_pages:
            images = content.get_images_for_vision()[:5]
            data = self._extract_from_images(images, schema)
        elif content.full_text:
            data = self._extract_from_text(content.full_text[:4000], schema)
        else:
            return ExtractionResult(document_type=doc_type, fields={}, confidence=0)

        return ExtractionResult(
            document_type=doc_type, fields=data,
            confidence=self._estimate_confidence(data, doc_type)
        )

    def _extract_from_images(self, images: list[dict], schema: str) -> dict:
        prompt = f"Extract: {schema}\nRespond ONLY with valid JSON. Use null for not found."
        content = [{"type": "text", "text": prompt}]
        for img in images:
            content.append({"type": "image_url", "image_url": {"url": f"data:image/png;base64,{img['base64']}"}})

        try:
            r = self.client.chat.completions.create(
                model="gpt-4o", messages=[{"role": "user", "content": content}],
                response_format={"type": "json_object"}, temperature=0, max_tokens=2000
            )
            return json.loads(r.choices[0].message.content)
        except Exception as e:
            if self.anthropic_client:
                return self._extract_anthropic_fallback(images, schema)
            raise

    def _extract_from_text(self, text: str, schema: str) -> dict:
        r = self.client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": f"Extract: {schema}\nValid JSON. null for not found.\n\n{text}"}],
            response_format={"type": "json_object"}, temperature=0, max_tokens=2000
        )
        return json.loads(r.choices[0].message.content)

    def _extract_anthropic_fallback(self, images: list[dict], schema: str) -> dict:
        content = []
        for img in images:
            content.append({"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": img["base64"]}})
        content.append({"type": "text", "text": f"Extract: {schema}\nValid JSON."})

        r = self.anthropic_client.messages.create(
            model="claude-3-5-sonnet-20241022", max_tokens=2000,
            messages=[{"role": "user", "content": content}]
        )
        return json.loads(r.content[0].text)

    def describe_for_rag(self, images: list[dict]) -> list[str]:
        descriptions = []
        for img in images[:10]:
            try:
                r = self.client.chat.completions.create(
                    model="gpt-4o-mini",
                    messages=[{"role": "user", "content": [
                        {"type": "text", "text": "Describe the content of this document image in 2-3 sentences."},
                        {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{img['base64']}"}}
                    ]}],
                    max_tokens=200, temperature=0
                )
                descriptions.append(f"[Page {img.get('page', '?')}] {r.choices[0].message.content}")
            except Exception as e:
                descriptions.append(f"[Page {img.get('page', '?')}] Error: {e}")
        return descriptions

    def _estimate_confidence(self, data: dict, doc_type: str) -> float:
        expected = {
            "invoice": ["date", "total", "vendor"],
            "contract": ["parties", "signing_date", "subject"],
            "manual": ["title", "sections"],
            "report": ["title", "sections"],
        }
        required = expected.get(doc_type, [])
        if not required:
            return 0.5
        found = sum(1 for f in required if data.get(f) is not None)
        return round(found / len(required), 2)

modules/rag_module.py (interface)

import logging
import os
from typing import Optional

import chromadb
from chromadb.utils import embedding_functions
from openai import OpenAI
from pydantic import BaseModel

logger = logging.getLogger(__name__)


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


class RAGModule:
    def __init__(self, collection_name: str = "document_analyzer", persist_directory: Optional[str] = None):
        self.openai_client = OpenAI()
        self.embedding_fn = embedding_functions.OpenAIEmbeddingFunction(
            api_key=os.getenv("OPENAI_API_KEY"), model_name="text-embedding-3-small"
        )
        self.chroma_client = (
            chromadb.PersistentClient(path=persist_directory) if persist_directory
            else chromadb.Client()
        )
        self.collection = self.chroma_client.get_or_create_collection(
            name=collection_name, embedding_function=self.embedding_fn
        )

    def index(self, doc_id: str, content, filename: str = "", document_type: str = "other", image_descriptions=None):
        if content.full_text:
            chunks = self._chunk(content.full_text, doc_id)
            if chunks:
                self.collection.add(
                    documents=[c["text"] for c in chunks],
                    ids=[c["id"] for c in chunks],
                    metadatas=[{"doc_id": doc_id, "content_type": "text", "filename": filename, "document_type": document_type} for c in chunks]
                )
        if image_descriptions:
            self.collection.add(
                documents=image_descriptions,
                ids=[f"{doc_id}_img_{i}" for i in range(len(image_descriptions))],
                metadatas=[{"doc_id": doc_id, "content_type": "image_description", "filename": filename, "document_type": document_type} for _ in image_descriptions]
            )

    def query(self, question: str, doc_id: Optional[str] = None, n_results: int = 5) -> QAResult:
        kwargs = {"query_texts": [question], "n_results": min(n_results, max(1, self.collection.count()))}
        if doc_id:
            kwargs["where"] = {"doc_id": {"$eq": doc_id}}

        results = self.collection.query(**kwargs)
        if not results["documents"] or not results["documents"][0]:
            return QAResult(question=question, answer="No relevant documents found.")

        docs = results["documents"][0]
        context = "\n\n".join(f"[Source {i+1}] {d}" for i, d in enumerate(docs))

        r = self.openai_client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": "Answer only based on the context. Cite sources with [Source N]."},
                {"role": "user", "content": f"Context:\n{context}\n\nQuestion: {question}"}
            ],
            max_tokens=500, temperature=0
        )

        sources = [{"text": d[:200], "index": i} for i, d in enumerate(docs)]
        return QAResult(question=question, answer=r.choices[0].message.content, sources=sources, chunks_used=len(docs))

    def list_documents(self) -> list[dict]:
        all_items = self.collection.get()
        docs = {}
        for meta in all_items.get("metadatas", []):
            did = meta.get("doc_id", "?")
            if did not in docs:
                docs[did] = {"doc_id": did, "filename": meta.get("filename", "?"), "document_type": meta.get("document_type", "?"), "chunks": 0}
            docs[did]["chunks"] += 1
        return list(docs.values())

    def _chunk(self, text: str, doc_id: str, size: int = 1500) -> list[dict]:
        chunks = []
        for i in range(0, len(text), size):
            chunk_text = text[i:i+size]
            if len(chunk_text.strip()) > 50:
                chunks.append({"id": f"{doc_id}_chunk_{len(chunks)}", "text": chunk_text})
        return chunks

modules/audio_module.py (interface)

import logging
import os
from openai import OpenAI

logger = logging.getLogger(__name__)


class AudioModule:
    def __init__(self, output_dir: str = "./audio_output"):
        self.client = OpenAI()
        self.output_dir = output_dir
        os.makedirs(output_dir, exist_ok=True)

    def generate_summary_audio(self, text: str, filename: str = "summary.mp3", voice: str = "nova", hd: bool = False) -> str:
        if len(text) > 4096:
            text = text[:4096]

        output_path = os.path.join(self.output_dir, filename)
        model = "tts-1-hd" if hd else "tts-1"

        response = self.client.audio.speech.create(model=model, voice=voice, input=text)
        response.stream_to_file(output_path)

        logger.info(f"Audio: {output_path} ({len(text)} chars)")
        return output_path

Demo Script

Run the complete system

"""
demo.py — Multimodal Document Analyzer demo
Run: python demo.py <file.pdf>
"""

import sys
import json
import logging

logging.basicConfig(level=logging.INFO)

from document_analyzer import DocumentAnalyzer


def main():
    if len(sys.argv) < 2:
        print("Usage: python demo.py <file.pdf|image.png> [question]")
        sys.exit(1)

    file_path = sys.argv[1]
    question = sys.argv[2] if len(sys.argv) > 2 else None

    print("=" * 60)
    print("  MULTIMODAL DOCUMENT ANALYZER — Demo")
    print("=" * 60)

    analyzer = DocumentAnalyzer()

    health = analyzer.health_check()
    print(f"\nHealth check: {health['status']}")
    if health["status"] != "ok":
        print(f"  Warning: {health}")

    print(f"\nAnalyzing: {file_path}")
    if question:
        print(f"Question: {question}")

    result = analyzer.analyze(
        file_path=file_path,
        question=question,
        extract_structured=True,
        generate_summary=True,
        generate_audio_summary=False,
        index_for_qa=True
    )

    print(f"\n{'=' * 60}")
    print(f"  RESULTS")
    print(f"{'=' * 60}")

    print(f"\nSuccess: {result['success']}")
    print(f"Doc ID: {result['doc_id']}")

    meta = result["metadata"]
    print(f"\n--- Metadata ---")
    print(f"  File: {meta['filename']}")
    print(f"  Type: {meta['file_type']}")
    print(f"  Pages: {meta['pages_processed']} (text: {meta['text_pages']}, image: {meta['image_pages']})")
    print(f"  Classification: {meta['document_type']}")
    print(f"  Indexed: {meta['indexed']}")
    print(f"  Latency: {meta['latency_seconds']}s")
    print(f"  Estimated cost: ${meta['estimated_cost_usd']}")

    if result.get("extracted_data"):
        print(f"\n--- Extracted Data ---")
        print(f"  Type: {result['extracted_data']['document_type']}")
        print(f"  Confidence: {result['extracted_data']['confidence']}")
        for key, value in result['extracted_data']['fields'].items():
            print(f"  {key}: {value}")

    if result.get("summary"):
        print(f"\n--- Summary ---")
        print(f"  {result['summary']}")

    if result.get("qa_result"):
        print(f"\n--- Q&A ---")
        print(f"  Question: {result['qa_result']['question']}")
        print(f"  Answer: {result['qa_result']['answer']}")
        if result['qa_result'].get('sources'):
            print(f"  Sources: {len(result['qa_result']['sources'])}")

    if result.get("errors"):
        print(f"\n--- Errors ---")
        for err in result['errors']:
            print(f"  ⚠ {err}")

    if result["success"] and result.get("metadata", {}).get("indexed"):
        print(f"\n--- Additional questions ---")
        follow_ups = [
            "What is the most important point of the document?",
            "Summarize in one sentence.",
        ]
        for q in follow_ups:
            try:
                answer = analyzer.ask(q, doc_id=result["doc_id"])
                print(f"  Q: {q}")
                print(f"  A: {answer['answer'][:200]}")
                print()
            except Exception as e:
                print(f"  Q: {q}")
                print(f"  Error: {e}")

    print(f"\n{'=' * 60}")
    print(f"  Demo completed")
    print(f"{'=' * 60}")


if __name__ == "__main__":
    main()

Test Cases

Unit tests

"""
test_document_analyzer.py — Document Analyzer tests
Run: python -m pytest test_document_analyzer.py -v
"""

import os
import json
import tempfile
from pathlib import Path

import pytest


def create_test_pdf(path: str, pages: int = 3, with_text: bool = True):
    import fitz
    doc = fitz.open()
    for i in range(pages):
        page = doc.new_page()
        if with_text:
            page.insert_text((72, 72), f"Test page {i+1}.\nSample content for testing.\nAdditional line with more text to exceed the threshold.")
    doc.save(path)
    doc.close()


class TestDocumentProcessor:
    def test_process_pdf_with_text(self):
        from modules.document_processor import DocumentProcessor

        with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as f:
            create_test_pdf(f.name, pages=2)
            processor = DocumentProcessor()
            result = processor.process(f.name)

            assert result.file_type == "pdf"
            assert result.total_pages == 2
            assert result.has_text_pages
            assert result.full_text is not None
            assert len(result.pages) >= 2

        os.unlink(f.name)

    def test_process_image(self):
        from modules.document_processor import DocumentProcessor
        from PIL import Image

        with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as f:
            img = Image.new("RGB", (100, 100), color="white")
            img.save(f.name)

            processor = DocumentProcessor()
            result = processor.process(f.name)

            assert result.file_type == "image"
            assert result.total_pages == 1
            assert result.has_image_pages
            assert len(result.get_images_for_vision()) == 1

        os.unlink(f.name)

    def test_unsupported_format(self):
        from modules.document_processor import DocumentProcessor

        processor = DocumentProcessor()
        with pytest.raises(ValueError, match="Unsupported"):
            processor.process("file.txt")


class TestDocumentAnalyzer:
    def test_validation_missing_file(self):
        from document_analyzer import DocumentAnalyzer

        analyzer = DocumentAnalyzer(chroma_persist_dir=None)
        result = analyzer.analyze("/does/not/exist.pdf")

        assert result["success"] is False
        assert "not found" in result["errors"][0].lower()

    def test_validation_unsupported_format(self):
        from document_analyzer import DocumentAnalyzer

        with tempfile.NamedTemporaryFile(suffix=".txt", delete=False) as f:
            f.write(b"test content")

        analyzer = DocumentAnalyzer(chroma_persist_dir=None)
        result = analyzer.analyze(f.name)

        assert result["success"] is False
        assert "not supported" in result["errors"][0].lower()

        os.unlink(f.name)

    def test_analyze_pdf_full_pipeline(self):
        from document_analyzer import DocumentAnalyzer

        with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as f:
            create_test_pdf(f.name, pages=2)

        analyzer = DocumentAnalyzer(chroma_persist_dir=None)
        result = analyzer.analyze(
            file_path=f.name,
            question="What is this document about?",
            extract_structured=True,
            generate_summary=True,
            generate_audio_summary=False,
            index_for_qa=True
        )

        assert result["success"] is True
        assert result["doc_id"] is not None
        assert result["metadata"]["pages_processed"] == 2
        assert result["summary"] is not None

        os.unlink(f.name)

    def test_health_check(self):
        from document_analyzer import DocumentAnalyzer

        analyzer = DocumentAnalyzer(chroma_persist_dir=None)
        health = analyzer.health_check()

        assert "status" in health
        assert "openai" in health

    def test_list_documents_empty(self):
        from document_analyzer import DocumentAnalyzer

        analyzer = DocumentAnalyzer(chroma_persist_dir=None)
        docs = analyzer.list_documents()

        assert isinstance(docs, list)

Final Deployment Checklist

Pre-deployment

• All modules have passing unit tests
• DocumentAnalyzer.analyze() processes text PDFs correctly
• DocumentAnalyzer.analyze() processes images correctly
• Structured extraction returns data for invoices
• Summary generates coherent text
• Indexing and Q&A return answers with sources
• Audio generates valid .mp3 files
• Health check returns the correct status
• Environment variables configured (.env)

Docker

• docker build completes without errors
• docker run brings up the service on port 8000
• curl http://localhost:8000/health returns {"status": "ok"}
• POST /analyze with a PDF returns a complete response
• Volumes mounted for chroma_data and audio_output

API

• POST /analyze accepts multipart files
• POST /ask answers questions about indexed documents
• GET /health returns the state of services
• GET /audio/{filename} serves audio files
• Errors return appropriate HTTP codes (400, 413, 500)
• Functional rate limiting

Production

• API keys as environment variables, never in code
• Structured logging in JSON
• Active cost monitoring
• Persistent RAG index (not lost on restart)
• Timeout configured for long requests
• CORS configured if there's a frontend

How to Run

Local execution (development)

cd document-analyzer

python -m venv venv
source venv/bin/activate

pip install -r requirements.txt

echo "OPENAI_API_KEY=sk-..." > .env

python demo.py invoice_example.pdf "What is the total?"

Execution with the REST API

uvicorn main:app --reload --host 0.0.0.0 --port 8000

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=false"

Execution with Docker

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

curl http://localhost:8000/health

curl -X POST "http://localhost:8000/analyze" \
  -F "file=@invoice_example.pdf" \
  -F "extract_structured=true"

Running tests

python -m pytest test_document_analyzer.py -v

python -m pytest test_document_analyzer.py -v --tb=short -x

Final Evaluation

Evaluation criteria

Completion levels

Final self-assessment

If you can check all these boxes, you completed the project:

• My DocumentAnalyzer.analyze() processes a PDF end-to-end without errors
• The classification correctly identifies the document type
• Structured extraction returns relevant data for invoices
• The summary captures the key points of the document
• I can ask questions and receive answers with sources
• The summary audio is generated and audible
• My REST API responds to POST /analyze and GET /health
• docker build completes without errors
• I understand the costs of each operation and how to optimize them
• I can explain the system architecture to a colleague

Exercises

Exercise 1: Batch analysis of multiple documents

Implement an analyze_batch() method in DocumentAnalyzer that receives a list of file paths and processes them sequentially, returning an aggregated summary with: total documents processed, successes/failures, total costs, and a unified RAG index for cross-document questions.

def analyze_batch(
    self,
    file_paths: list[str],
    index_for_qa: bool = True
) -> dict:
    results = []
    total_cost = 0
    success_count = 0
    error_count = 0

    for path in file_paths:
        logger.info(f"Processing {path} ({len(results)+1}/{len(file_paths)})")

        result = self.analyze(
            file_path=path,
            extract_structured=True,
            generate_summary=True,
            generate_audio_summary=False,
            index_for_qa=index_for_qa
        )

        results.append({
            "file": path,
            "doc_id": result["doc_id"],
            "success": result["success"],
            "document_type": result.get("metadata", {}).get("document_type", "?"),
            "pages": result.get("metadata", {}).get("pages_processed", 0),
            "cost": result.get("metadata", {}).get("estimated_cost_usd", 0),
            "errors": result.get("errors", [])
        })

        if result["success"]:
            success_count += 1
            total_cost += result.get("metadata", {}).get("estimated_cost_usd", 0)
        else:
            error_count += 1

    return {
        "total_documents": len(file_paths),
        "successful": success_count,
        "failed": error_count,
        "total_cost_usd": round(total_cost, 4),
        "indexed_for_qa": index_for_qa,
        "results": results,
        "rag_stats": self.rag.get_stats() if hasattr(self.rag, 'get_stats') else {}
    }


analyzer = DocumentAnalyzer()

batch_result = analyzer.analyze_batch([
    "invoice_january.pdf",
    "invoice_february.pdf",
    "service_contract.pdf"
])

print(f"Processed: {batch_result['successful']}/{batch_result['total_documents']}")
print(f"Total cost: ${batch_result['total_cost_usd']}")

cross_qa = analyzer.ask("How much did we pay in total in January and February?")
print(f"Cross-document answer: {cross_qa['answer']}")

Exercise 2: Export results to JSON

Implement an export_results() method that exports all cached results to a JSON file, including metadata, extracted data and summaries. Useful for auditing and reporting. The file must be readable (indented) and handle special characters correctly.

import json
from datetime import datetime


def export_results(
    self,
    output_path: str = "analysis_results.json",
    include_raw: bool = False
) -> str:
    export_data = {
        "export_timestamp": datetime.now().isoformat(),
        "total_documents": len(self.results_cache),
        "documents": []
    }

    for doc_id, result in self.results_cache.items():
        doc_export = {
            "doc_id": doc_id,
            "filename": result.get("metadata", {}).get("filename", "?"),
            "document_type": result.get("metadata", {}).get("document_type", "?"),
            "success": result.get("success", False),
            "pages_processed": result.get("metadata", {}).get("pages_processed", 0),
            "latency_seconds": result.get("metadata", {}).get("latency_seconds", 0),
            "estimated_cost_usd": result.get("metadata", {}).get("estimated_cost_usd", 0),
        }

        if result.get("extracted_data"):
            doc_export["extracted_data"] = result["extracted_data"]

        if result.get("summary"):
            doc_export["summary"] = result["summary"]

        if result.get("qa_result"):
            doc_export["qa_result"] = {
                "question": result["qa_result"]["question"],
                "answer": result["qa_result"]["answer"]
            }

        if result.get("errors"):
            doc_export["errors"] = result["errors"]

        export_data["documents"].append(doc_export)

    total_cost = sum(
        d.get("estimated_cost_usd", 0) for d in export_data["documents"]
    )
    export_data["total_estimated_cost_usd"] = round(total_cost, 4)

    with open(output_path, "w", encoding="utf-8") as f:
        json.dump(export_data, f, ensure_ascii=False, indent=2)

    logger.info(f"Results exported to {output_path}")
    return output_path


analyzer = DocumentAnalyzer()

analyzer.analyze("invoice_march.pdf", question="Total?")
analyzer.analyze("annual_contract.pdf")

export_path = analyzer.export_results("analysis_results.json")
print(f"Exported to: {export_path}")

Final Summary

What you built

The Multimodal Document Analyzer is a complete system that:

• Processes PDFs and images, automatically detecting text vs scanned pages
• Classifies the document type (invoice, contract, manual, report)
• Extracts structured data by type, with typed Pydantic schemas
• Summarizes the content into executive key points
• Indexes text and image descriptions in ChromaDB for multimodal RAG
• Answers specific questions with cited sources
• Generates audio summaries with OpenAI TTS
• Exposes everything as a REST API with FastAPI, ready for Docker

Integrated technologies

Integrated guide modules

Next step

This project is your multimodal AI portfolio piece. You can:

1. Deploy it on a VPS or cloud provider
2. Add a frontend (Streamlit, Next.js) for upload and visualization
3. Extend it with more document types (receipts, medical forms)
4. Improve it with fine-tuned classification, better OCR, or advanced RAG
5. Share it as an open source project on GitHub

Additional Resources

1. FastAPI — Web framework
2. PyMuPDF — PDF processing
3. ChromaDB — Vector database
4. OpenAI API — Vision, Chat, TTS, Embeddings
5. Anthropic API — Claude as fallback
6. Docker — Containerization
7. Pydantic — Data validation
8. pytest — Testing in Python