3. Procesamiento de Documentos

Descripción

El primer componente del Document Analyzer es el DocumentProcessor: la capa que recibe un archivo crudo (PDF o imagen) y lo transforma en datos que los demás módulos pueden consumir. Sin este componente, el VisionAnalyzer no tiene imágenes que analizar, el RAGModule no tiene texto que indexar, y el Summarizer no tiene contenido que resumir.

Por qué importa: El procesamiento de documentos es la base de todo el pipeline. Un PDF puede tener páginas con texto seleccionable (generado digitalmente), páginas escaneadas (imágenes), o una mezcla de ambos. Detectar correctamente el tipo de cada página determina si se extrae texto directamente (rápido, gratis) o se envía a Vision API (lento, costoso). Un error aquí se propaga a todos los módulos downstream.

Conexión con el módulo: En el Módulo 3 construiste funciones individuales para extraer texto e imágenes. Aquí las integras en una clase DocumentProcessor que encapsula toda la lógica de procesamiento, maneja casos edge (PDFs mixtos, imágenes corruptas, archivos grandes), y produce un ProcessedDocument estandarizado que consume el resto del sistema.

Pipeline de Procesamiento

Flujo para PDFs

PDF → Abrir con PyMuPDF → Iterar páginas
    │
    ├─── Página con texto (>50 chars) → Extraer texto → PageContent(type="text")
    │
    ├─── Página escaneada (<50 chars) → Renderizar a imagen → Base64 → PageContent(type="image")
    │
    └─── Página con imágenes embebidas → Extraer imágenes → Base64 → PageContent(type="image")

ProcessedDocument con lista de PageContent + metadata

Flujo para imágenes

Imagen → Verificar formato → Leer bytes → Base64 → PageContent(type="image")

ProcessedDocument con 1 PageContent + metadata

Decisión: texto vs imagen por página

La heurística clave es: ¿la página tiene suficiente texto extraíble?

TEXT_THRESHOLD = 50  # caracteres mínimos para considerar página como "texto"

Si len(page.get_text().strip()) > TEXT_THRESHOLD, la página tiene texto nativo. Si no, es escaneada o es una imagen y necesita procesamiento visual.

Esta heurística funciona bien para el 95% de los casos. Los edge cases (páginas con solo números o con texto en imágenes embebidas) se manejan con lógica adicional.

Modelos de Datos

Modelos que produce el DocumentProcessor

from pydantic import BaseModel
from typing import Optional


class PageContent(BaseModel):
    page_number: int
    content_type: str  # "text" | "image"
    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  # "pdf" | "image"
    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

Estos modelos son el contrato entre DocumentProcessor y todos los módulos downstream. VisionAnalyzer consume las páginas con content_type="image". RAGModule consume las páginas con content_type="text" y las descripciones generadas por Vision para las imágenes.

Implementación: DocumentProcessor

Clase completa

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

import fitz

logger = logging.getLogger(__name__)

TEXT_THRESHOLD = 50
IMAGE_DPI_SCALE = 150 / 72  # 150 DPI para renderizado de páginas
SUPPORTED_IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".webp"}


class DocumentProcessor:
    def __init__(self, text_threshold: int = TEXT_THRESHOLD, dpi_scale: float = IMAGE_DPI_SCALE):
        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)
        elif path.suffix.lower() in SUPPORTED_IMAGE_EXTENSIONS:
            return self._process_image(file_path)
        else:
            raise ValueError(f"Formato no soportado: {path.suffix}")

    def _process_pdf(self, file_path: str) -> ProcessedDocument:
        doc = fitz.open(file_path)
        pages: list[PageContent] = []
        full_text_parts: list[str] = []
        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)
                    ))
                    full_text_parts.append(text)
                    text_count += 1
                else:
                    image_b64 = self._page_to_base64(page)
                    pages.append(PageContent(
                        page_number=i + 1,
                        content_type="image",
                        image_base64=image_b64,
                        image_size_bytes=len(image_b64) * 3 // 4
                    ))
                    image_count += 1

                embedded = self._extract_embedded_images(page, i + 1)
                pages.extend(embedded)
                image_count += len(embedded)

        finally:
            doc.close()

        full_text = "\n\n".join(full_text_parts) if full_text_parts else None

        return ProcessedDocument(
            file_path=file_path,
            file_type="pdf",
            total_pages=len(doc),
            pages=pages,
            full_text=full_text,
            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()

        page = PageContent(
            page_number=1,
            content_type="image",
            image_base64=b64,
            image_size_bytes=len(raw)
        )

        return ProcessedDocument(
            file_path=file_path,
            file_type="image",
            total_pages=1,
            pages=[page],
            full_text=None,
            has_text_pages=False,
            has_image_pages=True,
            text_page_count=0,
            image_page_count=1
        )

    def _page_to_base64(self, page: fitz.Page) -> str:
        mat = fitz.Matrix(self.dpi_scale, self.dpi_scale)
        pix = page.get_pixmap(matrix=mat, alpha=False)
        png_bytes = pix.tobytes("png")
        return base64.b64encode(png_bytes).decode()

    def _extract_embedded_images(
        self, page: fitz.Page, page_number: int, min_size: int = 10000
    ) -> list[PageContent]:
        images = []
        image_list = page.get_images(full=True)

        for img_index, img_info in enumerate(image_list):
            xref = img_info[0]
            try:
                base_image = page.parent.extract_image(xref)
                if base_image and len(base_image["image"]) > min_size:
                    b64 = base64.b64encode(base_image["image"]).decode()
                    images.append(PageContent(
                        page_number=page_number,
                        content_type="image",
                        image_base64=b64,
                        image_size_bytes=len(base_image["image"])
                    ))
            except Exception as e:
                logger.warning(f"Error extrayendo imagen {img_index} de página {page_number}: {e}")

        return images

Chunking para RAG

El texto extraído necesita dividirse en chunks para indexación en ChromaDB. Chunks demasiado grandes diluyen la relevancia; demasiado pequeños pierden contexto.

Estrategia de chunking

Texto completo → Dividir por párrafos → Si párrafo > CHUNK_SIZE: subdividir
                                       → Si párrafo < MIN_CHUNK: combinar con siguiente
                                       → Agregar overlap entre chunks

Implementación

class TextChunker:
    def __init__(
        self,
        chunk_size: int = 1500,
        chunk_overlap: int = 200,
        min_chunk_size: int = 100
    ):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
        self.min_chunk_size = min_chunk_size

    def chunk_text(self, text: str, doc_id: str) -> list[DocumentChunk]:
        paragraphs = [p.strip() for p in text.split("\n\n") if p.strip()]
        chunks: list[DocumentChunk] = []
        current_chunk = ""
        chunk_index = 0

        for paragraph in paragraphs:
            if len(current_chunk) + len(paragraph) + 2 <= self.chunk_size:
                current_chunk += ("\n\n" + paragraph if current_chunk else paragraph)
            else:
                if len(current_chunk) >= self.min_chunk_size:
                    chunks.append(self._make_chunk(current_chunk, doc_id, chunk_index))
                    chunk_index += 1

                if len(paragraph) > self.chunk_size:
                    sub_chunks = self._split_long_paragraph(paragraph, doc_id, chunk_index)
                    chunks.extend(sub_chunks)
                    chunk_index += len(sub_chunks)
                    current_chunk = ""
                else:
                    overlap_text = current_chunk[-self.chunk_overlap:] if current_chunk else ""
                    current_chunk = overlap_text + "\n\n" + paragraph if overlap_text else paragraph

        if len(current_chunk) >= self.min_chunk_size:
            chunks.append(self._make_chunk(current_chunk, doc_id, chunk_index))

        return chunks

    def _split_long_paragraph(self, text: str, doc_id: str, start_index: int) -> list[DocumentChunk]:
        chunks = []
        for i in range(0, len(text), self.chunk_size - self.chunk_overlap):
            chunk_text = text[i:i + self.chunk_size]
            if len(chunk_text) >= self.min_chunk_size:
                chunks.append(self._make_chunk(chunk_text, doc_id, start_index + len(chunks)))
        return chunks

    def _make_chunk(self, text: str, doc_id: str, index: int) -> DocumentChunk:
        return DocumentChunk(
            chunk_id=f"{doc_id}_chunk_{index}",
            doc_id=doc_id,
            text=text,
            page_number=0,
            content_type="text",
            metadata={"char_count": len(text), "chunk_index": index}
        )


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

Procesamiento de Documentos Mixtos

El problema

Un PDF financiero puede tener:

• Páginas 1-3: texto generado digitalmente (datos de cuenta, movimientos)
• Páginas 4-5: imágenes escaneadas de comprobantes

El DocumentProcessor ya maneja esto por diseño: itera página por página y clasifica cada una. Pero para downstream, necesitamos poder filtrar fácilmente:

class ProcessedDocument(BaseModel):
    # ... campos anteriores ...

    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_text_for_rag(self) -> str:
        return "\n\n".join(p.text for p in self.get_text_pages() if p.text)

    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
        ]

Optimización de Imágenes

Reducir tamaño antes de enviar a Vision API

Las imágenes renderizadas a 150 DPI pueden ser más grandes de lo necesario. Reducir resolución ahorra costos (OpenAI cobra por tokens de imagen, que dependen del tamaño):

from PIL import Image
import io


def optimize_image_for_vision(
    image_base64: str,
    max_dimension: int = 2048,
    quality: int = 85
) -> str:
    raw = base64.b64decode(image_base64)
    img = Image.open(io.BytesIO(raw))

    if max(img.size) > max_dimension:
        ratio = max_dimension / max(img.size)
        new_size = (int(img.width * ratio), int(img.height * ratio))
        img = img.resize(new_size, Image.LANCZOS)

    buffer = io.BytesIO()
    img.save(buffer, format="PNG", optimize=True, quality=quality)
    return base64.b64encode(buffer.getvalue()).decode()

Tabla de costos por resolución

Para documentos, 1024×1024 o 2048×2048 suele ser suficiente. Solo necesitas resolución mayor para documentos con texto muy pequeño.

Manejo de Errores

Errores comunes y soluciones

Implementación de manejo de errores

def process_safe(self, file_path: str) -> ProcessedDocument | dict:
    try:
        return self.process(file_path)
    except fitz.FileDataError:
        logger.error(f"PDF corrupto: {file_path}")
        return {"error": "PDF corrupto o no legible", "file": file_path}
    except MemoryError:
        logger.error(f"Memoria insuficiente procesando: {file_path}")
        return {"error": "Documento demasiado grande para procesar", "file": file_path}
    except Exception as e:
        logger.error(f"Error inesperado: {e}", exc_info=True)
        return {"error": f"Error de procesamiento: {str(e)}", "file": file_path}

Uso del DocumentProcessor

Ejemplo completo

processor = DocumentProcessor(text_threshold=50, dpi_scale=150/72)
chunker = TextChunker(chunk_size=1500, chunk_overlap=200)

result = processor.process("factura_marzo.pdf")

print(f"Tipo: {result.file_type}")
print(f"Páginas totales: {result.total_pages}")
print(f"Páginas con texto: {result.text_page_count}")
print(f"Páginas como imagen: {result.image_page_count}")

if result.full_text:
    chunks = chunker.chunk_text(result.full_text, doc_id="factura_001")
    print(f"Chunks generados: {len(chunks)}")
    for chunk in chunks[:3]:
        print(f"  - {chunk.chunk_id}: {len(chunk.text)} chars")

images = result.get_images_for_vision()
if images:
    print(f"Imágenes para Vision: {len(images)}")
    for img in images:
        print(f"  - Página {img['page']}: {len(img['base64'])} chars base64")

Output esperado

Tipo: pdf
Páginas totales: 3
Páginas con texto: 2
Páginas como imagen: 1
Chunks generados: 4
  - factura_001_chunk_0: 1423 chars
  - factura_001_chunk_1: 1387 chars
  - factura_001_chunk_2: 892 chars
  - factura_001_chunk_3: 456 chars
Imágenes para Vision: 1
  - Página 3: 48920 chars base64

Troubleshooting

"PyMuPDF no extrae texto de mi PDF"

Causa probable: El PDF es escaneado (las "letras" son píxeles, no texto).

Verificación:

doc = fitz.open("documento.pdf")
for i in range(len(doc)):
    text = doc[i].get_text()
    print(f"Página {i+1}: {len(text.strip())} caracteres")
doc.close()

Si todas las páginas tienen 0 o pocos caracteres, el PDF es escaneado. El DocumentProcessor ya maneja este caso: convierte las páginas a imágenes para Vision.

"Las imágenes extraídas son muy grandes"

Solución: Usar la función optimize_image_for_vision() antes de enviar a la API:

for page in result.get_image_pages():
    optimized = optimize_image_for_vision(page.image_base64, max_dimension=1024)
    page.image_base64 = optimized

"Error de memoria con PDFs grandes"

Solución: Procesar páginas en lotes:

def process_pdf_batched(self, file_path: str, batch_size: int = 10) -> ProcessedDocument:
    doc = fitz.open(file_path)
    all_pages = []

    for batch_start in range(0, len(doc), batch_size):
        batch_end = min(batch_start + batch_size, len(doc))
        for i in range(batch_start, batch_end):
            page = doc[i]
            text = page.get_text()
            if len(text.strip()) > self.text_threshold:
                all_pages.append(PageContent(
                    page_number=i + 1, content_type="text",
                    text=text, char_count=len(text)
                ))
            else:
                b64 = self._page_to_base64(page)
                all_pages.append(PageContent(
                    page_number=i + 1, content_type="image",
                    image_base64=b64
                ))

    doc.close()
    # ... construir ProcessedDocument ...

Ejercicios

Ejercicio 1: Chunking con metadata de página

Modifica el TextChunker para que cada chunk incluya el número de página de donde proviene. Necesitas pasar la información de página desde ProcessedDocument al chunker.

def chunk_pages(self, pages: list[PageContent], doc_id: str) -> list[DocumentChunk]:
    chunks: list[DocumentChunk] = []
    chunk_index = 0

    for page in pages:
        if page.content_type != "text" or not page.text:
            continue

        paragraphs = [p.strip() for p in page.text.split("\n\n") if p.strip()]
        current_chunk = ""

        for paragraph in paragraphs:
            if len(current_chunk) + len(paragraph) + 2 <= self.chunk_size:
                current_chunk += ("\n\n" + paragraph if current_chunk else paragraph)
            else:
                if len(current_chunk) >= self.min_chunk_size:
                    chunks.append(DocumentChunk(
                        chunk_id=f"{doc_id}_chunk_{chunk_index}",
                        doc_id=doc_id,
                        text=current_chunk,
                        page_number=page.page_number,
                        content_type="text",
                        metadata={"source_page": page.page_number}
                    ))
                    chunk_index += 1
                current_chunk = paragraph

        if len(current_chunk) >= self.min_chunk_size:
            chunks.append(DocumentChunk(
                chunk_id=f"{doc_id}_chunk_{chunk_index}",
                doc_id=doc_id,
                text=current_chunk,
                page_number=page.page_number,
                content_type="text",
                metadata={"source_page": page.page_number}
            ))
            chunk_index += 1

    return chunks

Ejercicio 2: Detección de OCR necesario

Implementa una función que analice un ProcessedDocument y retorne un reporte: cuántas páginas necesitan OCR (Vision), cuántas tienen texto directo, y una estimación de costo basada en la cantidad de páginas-imagen.

COST_PER_IMAGE_PAGE = 0.01  # costo estimado de Vision por imagen
COST_PER_TEXT_PAGE = 0.0001  # costo de procesamiento de texto


def analyze_processing_needs(doc: ProcessedDocument) -> dict:
    text_pages = doc.get_text_pages()
    image_pages = doc.get_image_pages()

    text_cost = len(text_pages) * COST_PER_TEXT_PAGE
    image_cost = len(image_pages) * COST_PER_IMAGE_PAGE
    total_cost = text_cost + image_cost

    return {
        "total_pages": doc.total_pages,
        "text_pages": len(text_pages),
        "image_pages_needing_ocr": len(image_pages),
        "estimated_cost_usd": round(total_cost, 4),
        "cost_breakdown": {
            "text_processing": round(text_cost, 4),
            "vision_ocr": round(image_cost, 4)
        },
        "recommendation": (
            "Solo texto — procesamiento económico"
            if len(image_pages) == 0
            else f"{len(image_pages)} páginas requieren Vision API — "
                 f"costo estimado ${image_cost:.3f}"
        )
    }


doc = processor.process("informe_financiero.pdf")
report = analyze_processing_needs(doc)
print(report)

Resumen

• El DocumentProcessor es la base del pipeline: transforma archivos crudos en ProcessedDocument estandarizado.
• Detecta tipo de contenido por página: texto nativo (extracción directa) vs escaneado (requiere Vision).
• El TextChunker divide texto en chunks con overlap para indexación RAG óptima.
• Las imágenes se optimizan antes de enviar a Vision API para reducir costos.
• El manejo de errores cubre PDFs corruptos, memoria insuficiente y formatos no soportados.
• El procesamiento por lotes permite manejar PDFs grandes sin agotar memoria.

Recursos Adicionales

1. PyMuPDF Documentation — Referencia completa
2. PyMuPDF Recipes — Recetas de extracción de texto
3. Pillow Documentation — Procesamiento de imágenes
4. Módulo 3 de esta guía — Base de procesamiento de documentos