8. Proyecto: Document Analyzer Multimodal

Descripción

Este es el proyecto final de la Guía de IA Multimodal. Integra todos los componentes que construiste en las cápsulas anteriores — DocumentProcessor, VisionAnalyzer, RAGModule, AudioModule — en una sola clase DocumentAnalyzer que orquesta el pipeline completo. Incluye: el código integrado y funcional, un script de demo que puedes ejecutar, test cases para validar cada componente, un checklist de deployment, y la evaluación final.

Lo que entregas: Un sistema completo que recibe un PDF o imagen, lo procesa, extrae datos estructurados, genera resumen, indexa para Q&A, responde preguntas, y opcionalmente genera audio — todo a través de una API REST o directamente como módulo Python.

Criterio de éxito: Si puedes ejecutar el demo script y obtener resultados correctos para un PDF de ejemplo, tu proyecto está completo. Si además puedes hacer docker build y ejecutar curl contra la API, estás production-ready.

Arquitectura Final

Clase DocumentAnalyzer: el orquestador

DocumentAnalyzer
├── analyze()          → Pipeline completo: procesar + clasificar + extraer + resumir + indexar + Q&A + audio
├── ask()              → Q&A sobre documentos ya indexados
├── get_document()     → Recuperar resultado de análisis previo
├── list_documents()   → Listar documentos procesados
└── health_check()     → Verificar estado de servicios

Flujo interno de analyze()

analyze(file_path, question, options)
    │
    ├── 1. Validar input
    │      └── Extensión, tamaño, páginas
    │
    ├── 2. DocumentProcessor.process()
    │      └── Texto/imágenes por página
    │
    ├── 3. VisionAnalyzer.classify()
    │      └── factura | contrato | manual | informe | otro
    │
    ├── 4. VisionAnalyzer.extract_structured() [si extract_structured=True]
    │      └── Datos tipados según clasificación
    │
    ├── 5. Summarizer [si generate_summary=True]
    │      └── Resumen ejecutivo en 3-5 puntos
    │
    ├── 6. RAGModule.index() [si index_for_qa=True]
    │      └── Texto + descripciones de imágenes → ChromaDB
    │
    ├── 7. RAGModule.query() [si hay pregunta]
    │      └── Respuesta con fuentes citadas
    │
    ├── 8. AudioModule.generate_summary_audio() [si generate_audio=True]
    │      └── Archivo .mp3
    │
    └── 9. Construir AnalyzeResponse con metadata y costos

Implementación Completa: DocumentAnalyzer

Clase principal

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 inicializado")

    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 procesando documento: {e}"],
                "metadata": {"latency_seconds": round(time.time() - start, 2)}
            }

        doc_type = "otro"
        try:
            doc_type = self.analyzer.classify(content)
            logger.info(f"[{doc_id}] Clasificado como: {doc_type}")
        except Exception as e:
            errors.append(f"Clasificación falló: {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}] Extracción: confianza {extraction.confidence}")
            except Exception as e:
                errors.append(f"Extracción falló: {e}")

        summary = None
        if generate_summary:
            try:
                summary = self._generate_summary(content)
                logger.info(f"[{doc_id}] Resumen generado: {len(summary)} chars")
            except Exception as e:
                errors.append(f"Resumen falló: {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}] Indexado en RAG")
            except Exception as e:
                errors.append(f"Indexación falló: {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 completado")
            except Exception as e:
                errors.append(f"Q&A falló: {e}")

        audio_path = None
        if generate_audio_summary and summary:
            try:
                audio_filename = f"{doc_id}_resumen.mp3"
                audio_path = self.audio.generate_summary_audio(
                    text=summary,
                    filename=audio_filename,
                    voice=audio_voice
                )
                logger.info(f"[{doc_id}] Audio generado: {audio_path}")
            except Exception as e:
                errors.append(f"Audio falló: {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 ["Archivo no encontrado"]

        if p.suffix.lower() not in SUPPORTED_EXTENSIONS:
            errors.append(f"Formato '{p.suffix}' no soportado")
            return errors

        if p.stat().st_size > MAX_FILE_SIZE_BYTES:
            errors.append(
                f"Archivo de {p.stat().st_size / 1024 / 1024:.1f} MB "
                f"excede el límite de {MAX_FILE_SIZE_BYTES / 1024 / 1024:.0f} MB"
            )

        if p.stat().st_size == 0:
            errors.append("Archivo vacío")

        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 "Documento sin contenido extraíble."

        r = self.openai_client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{
                "role": "user",
                "content": f"Resume este documento en 3-5 puntos clave:\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": "Sin texto disponible para responder.",
                "sources": [],
                "confidence": 0
            }

        r = self.openai_client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{
                "role": "user",
                "content": (
                    f"Contexto:\n{content.full_text[:6000]}\n\n"
                    f"Pregunta: {question}\n\n"
                    "Responde basándote solo en el contexto."
                )
            }],
            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  # clasificación + resumen
        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)

Módulos Individuales (Resumen)

Para que el DocumentAnalyzer funcione, necesitas los módulos en modules/. Aquí está la estructura y un resumen de cada uno. El código completo está en las cápsulas 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 (interfaz)

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 = {
    "factura": "fecha, numero_factura, proveedor, receptor, subtotal, impuestos, total, moneda, items",
    "contrato": "tipo_contrato, partes, fecha_firma, fecha_vigencia, objeto, monto, clausulas_clave",
    "manual": "titulo, autor, fecha, secciones, resumen_ejecutivo",
    "informe": "titulo, autor, fecha, secciones, resumen_ejecutivo",
}


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 "otro"

    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": "Clasifica: factura, contrato, manual, informe, otro. Solo el nombre."},
                {"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"Clasifica: factura, contrato, manual, informe, otro. Solo el nombre.\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, "contenido_principal, puntos_clave")

        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"Extrae: {schema}\nResponde SOLO JSON válido. Usa null para no encontrado."
        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"Extrae: {schema}\nJSON válido. null para no encontrado.\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"Extrae: {schema}\nJSON válido."})

        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 el contenido de esta imagen de documento en 2-3 oraciones."},
                        {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{img['base64']}"}}
                    ]}],
                    max_tokens=200, temperature=0
                )
                descriptions.append(f"[Página {img.get('page', '?')}] {r.choices[0].message.content}")
            except Exception as e:
                descriptions.append(f"[Página {img.get('page', '?')}] Error: {e}")
        return descriptions

    def _estimate_confidence(self, data: dict, doc_type: str) -> float:
        expected = {
            "factura": ["fecha", "total", "proveedor"],
            "contrato": ["partes", "fecha_firma", "objeto"],
            "manual": ["titulo", "secciones"],
            "informe": ["titulo", "secciones"],
        }
        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 (interfaz)

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 = "otro", 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 se encontraron documentos relevantes.")

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

        r = self.openai_client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": "Responde solo con base en el contexto. Cita fuentes con [Fuente N]."},
                {"role": "user", "content": f"Contexto:\n{context}\n\nPregunta: {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 (interfaz)

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 = "resumen.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

Ejecutar el sistema completo

"""
demo.py — Demo del Document Analyzer Multimodal
Ejecutar: python demo.py <archivo.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("Uso: python demo.py <archivo.pdf|imagen.png> [pregunta]")
        sys.exit(1)

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

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

    analyzer = DocumentAnalyzer()

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

    print(f"\nAnalizando: {file_path}")
    if question:
        print(f"Pregunta: {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"  RESULTADOS")
    print(f"{'=' * 60}")

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

    meta = result["metadata"]
    print(f"\n--- Metadata ---")
    print(f"  Archivo: {meta['filename']}")
    print(f"  Tipo: {meta['file_type']}")
    print(f"  Páginas: {meta['pages_processed']} (texto: {meta['text_pages']}, imagen: {meta['image_pages']})")
    print(f"  Clasificación: {meta['document_type']}")
    print(f"  Indexado: {meta['indexed']}")
    print(f"  Latencia: {meta['latency_seconds']}s")
    print(f"  Costo estimado: ${meta['estimated_cost_usd']}")

    if result.get("extracted_data"):
        print(f"\n--- Datos Extraídos ---")
        print(f"  Tipo: {result['extracted_data']['document_type']}")
        print(f"  Confianza: {result['extracted_data']['confidence']}")
        for key, value in result['extracted_data']['fields'].items():
            print(f"  {key}: {value}")

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

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

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

    if result["success"] and result.get("metadata", {}).get("indexed"):
        print(f"\n--- Preguntas adicionales ---")
        follow_ups = [
            "¿Cuál es el punto más importante del documento?",
            "Resume en una oración.",
        ]
        for q in follow_ups:
            try:
                answer = analyzer.ask(q, doc_id=result["doc_id"])
                print(f"  P: {q}")
                print(f"  R: {answer['answer'][:200]}")
                print()
            except Exception as e:
                print(f"  P: {q}")
                print(f"  Error: {e}")

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


if __name__ == "__main__":
    main()

Test Cases

Tests unitarios

"""
test_document_analyzer.py — Tests del Document Analyzer
Ejecutar: 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"Página {i+1} de prueba.\nContenido de ejemplo para testing.\nLínea adicional con más texto para superar el 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="no soportado"):
            processor.process("archivo.txt")


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

        analyzer = DocumentAnalyzer(chroma_persist_dir=None)
        result = analyzer.analyze("/no/existe.pdf")

        assert result["success"] is False
        assert "no encontrado" 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 "no soportado" 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="¿De qué trata este documento?",
            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)

Deployment Checklist Final

Pre-deployment

• Todos los módulos tienen tests unitarios que pasan
• DocumentAnalyzer.analyze() procesa PDFs de texto correctamente
• DocumentAnalyzer.analyze() procesa imágenes correctamente
• Extracción estructurada retorna datos para facturas
• Resumen genera texto coherente
• Indexación y Q&A retornan respuestas con fuentes
• Audio genera archivos .mp3 válidos
• Health check retorna status correcto
• Variables de entorno configuradas (.env)

Docker

• docker build completa sin errores
• docker run levanta el servicio en puerto 8000
• curl http://localhost:8000/health retorna {"status": "ok"}
• POST /analyze con PDF retorna respuesta completa
• Volúmenes montados para chroma_data y audio_output

API

• POST /analyze acepta archivos multipart
• POST /ask responde preguntas sobre documentos indexados
• GET /health retorna estado de servicios
• GET /audio/{filename} sirve archivos de audio
• Errores retornan códigos HTTP apropiados (400, 413, 500)
• Rate limiting funcional

Producción

• API keys como environment variables, nunca en código
• Logging estructurado en JSON
• Monitoring de costos activo
• Índice RAG persistente (no se pierde al reiniciar)
• Timeout configurado para requests largos
• CORS configurado si hay frontend

Cómo Ejecutar

Ejecución local (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 factura_ejemplo.pdf "¿Cuál es el total?"

Ejecución con API REST

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

curl -X POST "http://localhost:8000/analyze" \
  -F "file=@factura_ejemplo.pdf" \
  -F "question=¿Cuál es el total?" \
  -F "extract_structured=true" \
  -F "generate_summary=true" \
  -F "generate_audio_summary=false"

Ejecución con 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=@factura_ejemplo.pdf" \
  -F "extract_structured=true"

Ejecución de tests

python -m pytest test_document_analyzer.py -v

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

Evaluación Final

Criterios de evaluación

Niveles de completitud

Autoevaluación final

Si puedes marcar todos estos checkboxes, completaste el proyecto:

• Mi DocumentAnalyzer.analyze() procesa un PDF end-to-end sin errores
• La clasificación identifica correctamente el tipo de documento
• La extracción estructurada retorna datos relevantes para facturas
• El resumen captura los puntos clave del documento
• Puedo hacer preguntas y recibir respuestas con fuentes
• El audio del resumen se genera y es audible
• Mi API REST responde a POST /analyze y GET /health
• docker build completa sin errores
• Entiendo los costos de cada operación y cómo optimizarlos
• Puedo explicar la arquitectura del sistema a un colega

Ejercicios

Ejercicio 1: Análisis batch de múltiples documentos

Implementa un método analyze_batch() en DocumentAnalyzer que reciba una lista de rutas de archivo y los procese secuencialmente, retornando un resumen agregado con: total de documentos procesados, éxitos/fallos, costos totales, y un índice RAG unificado para hacer preguntas cross-document.

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"Procesando {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([
    "factura_enero.pdf",
    "factura_febrero.pdf",
    "contrato_servicios.pdf"
])

print(f"Procesados: {batch_result['successful']}/{batch_result['total_documents']}")
print(f"Costo total: ${batch_result['total_cost_usd']}")

cross_qa = analyzer.ask("¿Cuánto pagamos en total en enero y febrero?")
print(f"Respuesta cross-document: {cross_qa['answer']}")

Ejercicio 2: Exportar resultados a JSON

Implementa un método export_results() que exporte todos los resultados cacheados a un archivo JSON, incluyendo metadata, datos extraídos y resúmenes. Útil para auditoría y reportes. El archivo debe ser legible (indentado) y manejar correctamente caracteres especiales en español.

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"Resultados exportados a {output_path}")
    return output_path


analyzer = DocumentAnalyzer()

analyzer.analyze("factura_marzo.pdf", question="¿Total?")
analyzer.analyze("contrato_anual.pdf")

export_path = analyzer.export_results("resultados_analisis.json")
print(f"Exportado a: {export_path}")

Resumen Final

Qué construiste

El Document Analyzer Multimodal es un sistema completo que:

• Procesa PDFs y imágenes, detectando automáticamente texto vs páginas escaneadas
• Clasifica el tipo de documento (factura, contrato, manual, informe)
• Extrae datos estructurados según el tipo, con schemas tipados en Pydantic
• Resume el contenido en puntos clave ejecutivos
• Indexa texto y descripciones de imágenes en ChromaDB para RAG multimodal
• Responde preguntas específicas con fuentes citadas
• Genera resúmenes en audio con TTS de OpenAI
• Expone todo como API REST con FastAPI, lista para Docker

Tecnologías integradas

Módulos de la guía integrados

Siguiente paso

Este proyecto es tu portfolio piece de IA multimodal. Puedes:

1. Desplegarlo en un VPS o cloud provider
2. Agregar un frontend (Streamlit, Next.js) para upload y visualización
3. Extenderlo con más tipos de documento (recibos, formularios médicos)
4. Mejorarlo con fine-tuning de clasificación, OCR mejorado, o RAG avanzado
5. Compartirlo como proyecto open source en GitHub

Recursos Adicionales

1. FastAPI — Framework web
2. PyMuPDF — Procesamiento de PDFs
3. ChromaDB — Base de datos vectorial
4. OpenAI API — Vision, Chat, TTS, Embeddings
5. Anthropic API — Claude como fallback
6. Docker — Containerización
7. Pydantic — Validación de datos
8. pytest — Testing en Python