Módulo 3: Comprensión de Documentos

8. Proyecto: Document Extractor

Siguiente →

Descripción

Este proyecto cierra el Módulo 3 construyendo un Document Extractor completo: un sistema que recibe un PDF o imagen de documento, detecta su tipo, extrae texto con la técnica apropiada (PyMuPDF, Tesseract, Vision), convierte la información en datos estructurados con Pydantic, y maneja documentos largos con chunking. Integra todas las cápsulas del módulo en un pipeline funcional.

No es un script aislado. Es el patrón que usan sistemas de procesamiento documental en producción: un router que decide la ruta de extracción según el tipo de documento, y un parser que convierte texto libre en datos tipados y validados.

Por qué importa: En producción, los documentos llegan en cualquier formato — PDFs con texto, PDFs escaneados, fotografías de facturas. Sin un sistema que detecte el tipo y aplique la técnica correcta, terminas con pipelines frágiles que fallan con el primer documento inesperado.

Conexión con el módulo: Cada componente viene de una cápsula: PDFs (02), OCR/Vision (03), imágenes de documentos (04), Pydantic (05), chunking (06), errores (07).

Conexión con la guía: Este extractor es la base del Document Analyzer del Módulo 8. Allí le agregarás RAG para Q&A sobre el contenido extraído, y opcionalmente TTS para resúmenes hablados.

Especificaciones Técnicas

Input

El extractor acepta dos argumentos:

ParámetroTipoDescripción
file_pathstrRuta a PDF o imagen (JPG, PNG, WEBP)
schema_typestrTipo de schema: "invoice", "receipt", "contract", "auto"

Output

@dataclass
class ExtractionResult:
    success: bool                    # Si la extracción fue exitosa
    data: dict | None                # Datos estructurados validados por Pydantic
    raw_text: str                    # Texto crudo extraído del documento
    pages_processed: int             # Número de páginas procesadas
    method: str                      # "pymupdf" | "tesseract" | "vision"
    document_type: str               # "pdf_text" | "pdf_scanned" | "image"
    schema_used: str                 # "invoice" | "receipt" | "contract"
    cost_usd: float                  # Costo estimado de la extracción
    confidence: float                # 0.0-1.0 confianza en la extracción
    errors: list[str]                # Errores encontrados durante el proceso
    latency_seconds: float           # Tiempo total de ejecución

Requisitos funcionales

  1. Detectar tipo de documento: PDF con texto, PDF escaneado, imagen
  2. Extraer texto con la técnica óptima según el tipo detectado
  3. Definir schemas Pydantic para facturas, recibos y contratos
  4. Extraer datos estructurados enviando texto/imagen al LLM con el schema
  5. Manejar documentos largos con chunking por páginas y merge de resultados
  6. Reportar costo, confianza y método usado en cada extracción

Paso 1: Configuración y Estructuras de Datos

Definimos los enums, dataclasses y configuración base. Separar constantes de lógica permite cambiar precios o métodos sin tocar el pipeline.

from dataclasses import dataclass, field
from enum import Enum
from pathlib import Path
from typing import Any, Optional
import base64
import json
import time

import fitz
from openai import OpenAI
from pydantic import BaseModel, Field, ValidationError


class DocumentType(Enum):
    PDF_TEXT = "pdf_text"
    PDF_SCANNED = "pdf_scanned"
    IMAGE = "image"
    UNKNOWN = "unknown"


class ExtractionMethod(Enum):
    PYMUPDF = "pymupdf"
    TESSERACT = "tesseract"
    VISION = "vision"


SUPPORTED_EXTENSIONS = {
    ".pdf": "pdf",
    ".jpg": "image", ".jpeg": "image",
    ".png": "image", ".webp": "image",
}

METHOD_COSTS_PER_PAGE = {
    ExtractionMethod.PYMUPDF: 0.0,
    ExtractionMethod.TESSERACT: 0.0,
    ExtractionMethod.VISION: 0.003,
}

MIN_TEXT_CHARS_FOR_DIGITAL = 50


@dataclass
class ExtractionResult:
    success: bool = False
    data: dict | None = None
    raw_text: str = ""
    pages_processed: int = 0
    method: str = ""
    document_type: str = ""
    schema_used: str = ""
    cost_usd: float = 0.0
    confidence: float = 0.0
    errors: list[str] = field(default_factory=list)
    latency_seconds: float = 0.0


client = OpenAI()

DocumentType codifica los tres casos que el sistema debe manejar. ExtractionMethod determina qué técnica usar. Los costos por página son aproximados — Vision API cobra por tokens de imagen, pero para estimación rápida usamos un promedio por página.

Paso 2: Detectar Tipo de Documento

La detección sigue tres reglas: si es imagen, es imagen. Si es PDF, extraemos texto con PyMuPDF — si tiene más de 50 caracteres por página en promedio, es PDF digital; si no, es PDF escaneado.

def detect_document_type(file_path: str) -> DocumentType:
    """Clasifica el documento como PDF digital, PDF escaneado, o imagen."""
    path = Path(file_path)

    if not path.exists():
        raise FileNotFoundError(f"Archivo no encontrado: {file_path}")

    suffix = path.suffix.lower()
    if suffix not in SUPPORTED_EXTENSIONS:
        raise ValueError(f"Formato no soportado: {suffix}")

    if SUPPORTED_EXTENSIONS[suffix] == "image":
        return DocumentType.IMAGE

    doc = fitz.open(file_path)
    total_text = ""
    for page_num in range(len(doc)):
        total_text += doc[page_num].get_text()
    doc.close()

    avg_chars_per_page = len(total_text.strip()) / max(len(doc), 1)
    if avg_chars_per_page > MIN_TEXT_CHARS_FOR_DIGITAL:
        return DocumentType.PDF_TEXT

    return DocumentType.PDF_SCANNED

Paso 3: Extraer Texto

Tres métodos de extracción. PyMuPDF para PDFs digitales (gratis, rápido). Tesseract para OCR local cuando Vision no es opción. Vision API para máxima calidad en documentos escaneados o imágenes.

def extract_text_pymupdf(file_path: str) -> tuple[str, int]:
    """Extrae texto de PDF digital con PyMuPDF. Retorna (texto, páginas)."""
    doc = fitz.open(file_path)
    pages = []
    for page_num in range(len(doc)):
        page_text = doc[page_num].get_text()
        if page_text.strip():
            pages.append(page_text)
    doc.close()
    return "\n\n".join(pages), len(pages)


def extract_text_tesseract(file_path: str) -> tuple[str, int]:
    """Extrae texto con Tesseract OCR. Convierte PDF a imágenes primero."""
    import pytesseract
    from pdf2image import convert_from_path
    from PIL import Image

    path = Path(file_path)
    if path.suffix.lower() == ".pdf":
        images = convert_from_path(file_path, dpi=200)
    else:
        images = [Image.open(file_path)]

    texts = []
    for img in images:
        text = pytesseract.image_to_string(img, lang="spa+eng")
        texts.append(text)

    return "\n\n".join(texts), len(images)


def extract_text_vision(file_path: str) -> tuple[str, int]:
    """Extrae texto enviando imágenes a GPT-4 Vision."""
    path = Path(file_path)
    images_b64 = []

    if path.suffix.lower() == ".pdf":
        doc = fitz.open(file_path)
        for page_num in range(min(len(doc), 10)):
            page = doc[page_num]
            mat = fitz.Matrix(150 / 72, 150 / 72)
            pix = page.get_pixmap(matrix=mat, alpha=False)
            b64 = base64.b64encode(pix.tobytes("png")).decode()
            images_b64.append(b64)
        doc.close()
    else:
        with open(file_path, "rb") as f:
            b64 = base64.b64encode(f.read()).decode()
        images_b64.append(b64)

    prompt = "Extrae TODO el texto visible en este documento. Mantén la estructura original (títulos, listas, tablas). Responde solo con el texto extraído."

    content = [{"type": "text", "text": prompt}]
    for b64 in images_b64:
        content.append({
            "type": "image_url",
            "image_url": {"url": f"data:image/png;base64,{b64}"}
        })

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": content}],
        temperature=0,
        max_tokens=4096,
    )

    return response.choices[0].message.content, len(images_b64)


def select_and_extract(
    file_path: str, doc_type: DocumentType
) -> tuple[str, int, ExtractionMethod]:
    """Selecciona método de extracción según tipo de documento."""
    if doc_type == DocumentType.PDF_TEXT:
        text, pages = extract_text_pymupdf(file_path)
        return text, pages, ExtractionMethod.PYMUPDF

    if doc_type == DocumentType.PDF_SCANNED:
        text, pages = extract_text_vision(file_path)
        return text, pages, ExtractionMethod.VISION

    if doc_type == DocumentType.IMAGE:
        text, pages = extract_text_vision(file_path)
        return text, pages, ExtractionMethod.VISION

    raise ValueError(f"Tipo de documento no reconocido: {doc_type}")

select_and_extract es el router central. PDFs digitales van a PyMuPDF (costo cero). Escaneados e imágenes van a Vision. Si quisieras un fallback a Tesseract por costo, cambias una línea aquí.

Paso 4: Schemas Pydantic

Tres schemas para los tipos de documento más comunes. El schema registry mapea nombres a clases Pydantic y permite auto-detección por keywords.

class InvoiceItem(BaseModel):
    descripcion: str = Field(default="", description="Descripción del producto/servicio")
    cantidad: float = Field(default=1.0, description="Cantidad")
    precio_unitario: float = Field(default=0.0, description="Precio por unidad")
    monto: float = Field(default=0.0, description="Monto total del item")


class InvoiceSchema(BaseModel):
    fecha: Optional[str] = Field(default=None, description="Fecha en formato YYYY-MM-DD")
    numero_factura: Optional[str] = Field(default=None, description="Número de factura")
    proveedor: Optional[str] = Field(default=None, description="Nombre del proveedor/emisor")
    cliente: Optional[str] = Field(default=None, description="Nombre del cliente/receptor")
    subtotal: Optional[float] = Field(default=None, description="Subtotal antes de impuestos")
    impuestos: Optional[float] = Field(default=None, description="Monto de impuestos")
    total: Optional[float] = Field(default=None, description="Total a pagar")
    moneda: Optional[str] = Field(default=None, description="Moneda (USD, EUR, MXN)")
    items: list[InvoiceItem] = Field(default_factory=list, description="Lista de items")


class ReceiptItem(BaseModel):
    descripcion: str = Field(default="", description="Nombre del producto")
    cantidad: float = Field(default=1.0)
    precio: float = Field(default=0.0)


class ReceiptSchema(BaseModel):
    fecha: Optional[str] = Field(default=None, description="Fecha YYYY-MM-DD")
    comercio: Optional[str] = Field(default=None, description="Nombre del comercio")
    direccion: Optional[str] = Field(default=None, description="Dirección del comercio")
    items: list[ReceiptItem] = Field(default_factory=list)
    subtotal: Optional[float] = None
    impuesto: Optional[float] = None
    total: Optional[float] = None
    metodo_pago: Optional[str] = Field(default=None, description="Efectivo, tarjeta, etc.")


class ContractSchema(BaseModel):
    titulo: Optional[str] = Field(default=None, description="Título del contrato")
    fecha: Optional[str] = Field(default=None, description="Fecha de firma YYYY-MM-DD")
    partes: list[str] = Field(default_factory=list, description="Partes involucradas")
    objeto: Optional[str] = Field(default=None, description="Objeto del contrato")
    vigencia: Optional[str] = Field(default=None, description="Período de vigencia")
    monto: Optional[float] = Field(default=None, description="Monto del contrato")
    clausulas_clave: list[str] = Field(
        default_factory=list, description="Cláusulas principales resumidas"
    )


SCHEMA_REGISTRY: dict[str, type[BaseModel]] = {
    "invoice": InvoiceSchema,
    "receipt": ReceiptSchema,
    "contract": ContractSchema,
}

SCHEMA_KEYWORDS: dict[str, list[str]] = {
    "invoice": ["factura", "invoice", "nº factura", "subtotal", "iva", "proveedor"],
    "receipt": ["ticket", "recibo", "receipt", "comercio", "cajero", "cambio"],
    "contract": ["contrato", "contract", "cláusula", "vigencia", "partes", "firmante"],
}


def detect_schema_type(text: str) -> str:
    """Auto-detecta el tipo de schema analizando palabras clave en el texto."""
    text_lower = text.lower()
    scores = {}
    for schema_name, keywords in SCHEMA_KEYWORDS.items():
        score = sum(1 for kw in keywords if kw in text_lower)
        scores[schema_name] = score

    best = max(scores, key=scores.get)
    if scores[best] == 0:
        return "invoice"
    return best

Cada schema usa Field(default=None) — un documento puede no tener todos los campos, y extraemos lo que hay sin fallar. La auto-detección por keywords es simple pero efectiva; en producción podrías reemplazarla con un clasificador LLM.

Paso 5: Extracción Estructurada con LLM

Enviamos el texto extraído al LLM junto con el schema JSON. El LLM devuelve JSON que parseamos y validamos con Pydantic.

def extract_structured_data(
    text: str,
    schema_type: str,
    images_b64: list[str] | None = None,
) -> tuple[dict, float]:
    """Extrae datos estructurados usando LLM + schema Pydantic.

    Returns:
        (datos_validados, confianza)
    """
    schema_class = SCHEMA_REGISTRY.get(schema_type)
    if not schema_class:
        raise ValueError(f"Schema no registrado: {schema_type}")

    schema_json = schema_class.model_json_schema()
    schema_str = json.dumps(schema_json, indent=2, ensure_ascii=False)

    prompt = f"""Extrae los datos de este documento según el schema proporcionado.

SCHEMA (JSON Schema):
{schema_str}

REGLAS:
- Responde ÚNICAMENTE con JSON válido que cumpla el schema
- Si un campo no está presente en el documento, usa null
- Fechas en formato YYYY-MM-DD
- Montos como números (sin símbolos de moneda)
- Si hay items/líneas, extrae todos los que encuentres

DOCUMENTO:
{text[:6000]}"""

    content: list[dict[str, Any]] = [{"type": "text", "text": prompt}]

    if images_b64:
        for b64 in images_b64[:5]:
            content.append({
                "type": "image_url",
                "image_url": {"url": f"data:image/png;base64,{b64}"}
            })

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": content}],
        response_format={"type": "json_object"},
        temperature=0,
    )

    raw_json = json.loads(response.choices[0].message.content)

    try:
        validated = schema_class(**raw_json)
        data = validated.model_dump()
        confidence = _calculate_field_confidence(data)
        return data, confidence
    except ValidationError as e:
        data = _safe_partial_parse(raw_json, schema_class)
        confidence = _calculate_field_confidence(data) * 0.7
        return data, confidence


def _safe_partial_parse(raw: dict, schema_class: type[BaseModel]) -> dict:
    """Intenta parsear parcialmente cuando la validación completa falla."""
    clean = {}
    for field_name, field_info in schema_class.model_fields.items():
        if field_name in raw:
            try:
                clean[field_name] = raw[field_name]
            except (TypeError, ValueError):
                clean[field_name] = None
        else:
            clean[field_name] = None
    return clean


def _calculate_field_confidence(data: dict) -> float:
    """Calcula confianza basada en proporción de campos no-nulos."""
    if not data:
        return 0.0
    total = len(data)
    filled = sum(1 for v in data.values() if v is not None and v != "" and v != [])
    return round(filled / max(total, 1), 2)

Paso 6: Manejo de Documentos Largos

Para PDFs de más de 5 páginas, procesamos por chunks y mergeamos los resultados. _safe_partial_parse actúa como fallback: si el LLM devuelve JSON que no cumple el schema, extraemos lo posible en lugar de perder la extracción.

CHUNK_SIZE_PAGES = 5


def extract_long_document(
    file_path: str,
    schema_type: str,
) -> tuple[str, list[dict], int]:
    """Extrae texto de documento largo en chunks por páginas.

    Returns:
        (texto_completo, chunks_data, total_pages)
    """
    doc = fitz.open(file_path)
    total_pages = len(doc)
    all_text_parts = []
    chunks_data = []

    for start in range(0, total_pages, CHUNK_SIZE_PAGES):
        end = min(start + CHUNK_SIZE_PAGES, total_pages)
        chunk_text = ""
        for page_num in range(start, end):
            chunk_text += doc[page_num].get_text() + "\n\n"

        all_text_parts.append(chunk_text)

        if chunk_text.strip():
            data, conf = extract_structured_data(chunk_text, schema_type)
            chunks_data.append({"pages": f"{start+1}-{end}", "data": data, "confidence": conf})

    doc.close()
    full_text = "\n\n".join(all_text_parts)
    return full_text, chunks_data, total_pages


def merge_chunk_results(chunks_data: list[dict], schema_type: str) -> tuple[dict, float]:
    """Combina resultados de múltiples chunks en un solo resultado.

    Estrategia: para campos escalares toma el primer valor no-nulo.
    Para listas (items, cláusulas) concatena todos.
    """
    if not chunks_data:
        return {}, 0.0

    if len(chunks_data) == 1:
        return chunks_data[0]["data"], chunks_data[0]["confidence"]

    schema_class = SCHEMA_REGISTRY[schema_type]
    merged = {}
    list_fields = set()

    for field_name, field_info in schema_class.model_fields.items():
        if hasattr(field_info.annotation, "__origin__") and field_info.annotation.__origin__ is list:
            list_fields.add(field_name)
            merged[field_name] = []
        else:
            merged[field_name] = None

    for chunk in chunks_data:
        data = chunk["data"]
        for key, value in data.items():
            if key in list_fields and isinstance(value, list):
                merged[key].extend(value)
            elif merged.get(key) is None and value is not None:
                merged[key] = value

    avg_confidence = sum(c["confidence"] for c in chunks_data) / len(chunks_data)
    return merged, round(avg_confidence, 2)

El merge prioriza el primer valor no-nulo para campos escalares (fecha, total). Para listas como items o cláusulas, concatena todos los chunks.

Paso 7: Función Principal

extract_document() integra todos los pasos en un pipeline limpio. Es el único punto de entrada que necesita el usuario.

def extract_document(
    file_path: str,
    schema_type: str = "auto",
) -> ExtractionResult:
    """Pipeline completo de extracción de documentos.

    Args:
        file_path: Ruta a PDF o imagen
        schema_type: "invoice", "receipt", "contract", o "auto" para detectar

    Returns:
        ExtractionResult con datos, metadata y métricas
    """
    start = time.time()
    result = ExtractionResult()

    try:
        doc_type = detect_document_type(file_path)
        result.document_type = doc_type.value

        text, pages, method = select_and_extract(file_path, doc_type)
        result.raw_text = text
        result.pages_processed = pages
        result.method = method.value
        result.cost_usd = METHOD_COSTS_PER_PAGE[method] * pages

        if schema_type == "auto":
            schema_type = detect_schema_type(text)
        result.schema_used = schema_type

        if pages > CHUNK_SIZE_PAGES and doc_type == DocumentType.PDF_TEXT:
            full_text, chunks_data, total_pages = extract_long_document(
                file_path, schema_type
            )
            result.raw_text = full_text
            result.pages_processed = total_pages
            data, confidence = merge_chunk_results(chunks_data, schema_type)
        else:
            data, confidence = extract_structured_data(text, schema_type)

        result.data = data
        result.confidence = confidence
        result.success = True

    except FileNotFoundError as e:
        result.errors.append(f"Archivo no encontrado: {e}")
    except ValueError as e:
        result.errors.append(f"Error de formato: {e}")
    except Exception as e:
        result.errors.append(f"Error inesperado: {type(e).__name__}: {e}")

    result.latency_seconds = round(time.time() - start, 2)
    return result

Demo: Uso Completo

def print_result(result: ExtractionResult):
    """Imprime resultado de extracción de forma legible."""
    status = "ÉXITO" if result.success else "ERROR"
    print(f"\n{'='*60}")
    print(f"  Estado:        {status}")
    print(f"  Tipo doc:      {result.document_type}")
    print(f"  Método:        {result.method}")
    print(f"  Schema:        {result.schema_used}")
    print(f"  Páginas:       {result.pages_processed}")
    print(f"  Confianza:     {result.confidence:.0%}")
    print(f"  Costo est.:    ${result.cost_usd:.4f}")
    print(f"  Latencia:      {result.latency_seconds}s")

    if result.data:
        print(f"  Datos extraídos:")
        for key, value in result.data.items():
            if isinstance(value, list) and len(value) > 2:
                print(f"    {key}: [{len(value)} items]")
            else:
                print(f"    {key}: {value}")

    if result.errors:
        print(f"  Errores:")
        for err in result.errors:
            print(f"    - {err}")
    print(f"{'='*60}")


# --- Ejemplo 1: Factura en PDF digital ---
result = extract_document("factura_digital.pdf", schema_type="invoice")
print_result(result)

# --- Ejemplo 2: Recibo escaneado (imagen) ---
result = extract_document("recibo_foto.jpg", schema_type="receipt")
print_result(result)

# --- Ejemplo 3: Contrato largo con auto-detección ---
result = extract_document("contrato_20paginas.pdf", schema_type="auto")
print_result(result)

# --- Ejemplo 4: Acceso directo al JSON ---
result = extract_document("factura.pdf")
if result.success:
    print(json.dumps(result.data, indent=2, ensure_ascii=False))

Salida esperada (factura digital):

============================================================
  Estado:        ÉXITO
  Tipo doc:      pdf_text
  Método:        pymupdf
  Schema:        invoice
  Páginas:       1
  Confianza:     89%
  Costo est.:    $0.0030
  Latencia:      1.84s
  Datos extraídos:
    fecha: 2024-03-15
    numero_factura: INV-2024-0847
    proveedor: Acme Technologies S.A.
    total: 5220.0
    items: [3 items]
============================================================

Extensión 1: Batch Processing

Procesa múltiples documentos con tracking de progreso, costos acumulados, y resumen estadístico.

@dataclass
class BatchResult:
    total: int = 0
    successful: int = 0
    failed: int = 0
    total_cost_usd: float = 0.0
    total_pages: int = 0
    avg_confidence: float = 0.0
    by_type: dict = field(default_factory=dict)
    by_method: dict = field(default_factory=dict)
    results: list[ExtractionResult] = field(default_factory=list)


def extract_batch(
    file_paths: list[str],
    schema_type: str = "auto",
    budget_usd: float | None = None,
) -> BatchResult:
    """Procesa múltiples documentos con tracking de progreso."""
    batch = BatchResult()
    confidences = []

    for i, path in enumerate(file_paths):
        print(f"  [{i+1}/{len(file_paths)}] Procesando: {Path(path).name}...")

        if budget_usd and batch.total_cost_usd >= budget_usd:
            fail = ExtractionResult()
            fail.errors.append(f"Presupuesto agotado: ${batch.total_cost_usd:.4f}/{budget_usd}")
            batch.results.append(fail)
            batch.failed += 1
            batch.total += 1
            continue

        result = extract_document(path, schema_type=schema_type)
        batch.results.append(result)
        batch.total += 1

        if result.success:
            batch.successful += 1
            batch.total_cost_usd += result.cost_usd
            batch.total_pages += result.pages_processed
            confidences.append(result.confidence)

            batch.by_type[result.document_type] = batch.by_type.get(result.document_type, 0) + 1
            batch.by_method[result.method] = batch.by_method.get(result.method, 0) + 1
        else:
            batch.failed += 1

    batch.avg_confidence = round(sum(confidences) / len(confidences), 2) if confidences else 0.0
    return batch


def print_batch_report(batch: BatchResult):
    """Imprime reporte de batch processing."""
    print(f"\n{'='*60}")
    print(f"  REPORTE DE BATCH")
    print(f"{'='*60}")
    print(f"  Total documentos:  {batch.total}")
    print(f"  Exitosos:          {batch.successful}")
    print(f"  Fallidos:          {batch.failed}")
    print(f"  Total páginas:     {batch.total_pages}")
    print(f"  Costo total:       ${batch.total_cost_usd:.4f}")
    print(f"  Confianza prom.:   {batch.avg_confidence:.0%}")

    if batch.by_type:
        print(f"  Por tipo:")
        for doc_type, count in batch.by_type.items():
            print(f"    {doc_type}: {count}")

    if batch.by_method:
        print(f"  Por método:")
        for method, count in batch.by_method.items():
            print(f"    {method}: {count}")
    print(f"{'='*60}")

Extensión 2: Confidence Scoring Detallado

Un scoring más granular que analiza la calidad de cada campo extraído, no solo la proporción de campos llenos.

@dataclass
class FieldScore:
    field_name: str
    present: bool
    plausible: bool
    score: float


def score_extraction(data: dict, schema_type: str) -> tuple[float, list[FieldScore]]:
    """Evalúa la calidad de la extracción campo por campo.

    Checks:
    - Presencia: el campo tiene valor no-nulo
    - Plausibilidad: el valor tiene formato esperado
    """
    field_scores = []

    plausibility_checks = {
        "fecha": lambda v: bool(v and len(str(v)) == 10 and "-" in str(v)),
        "total": lambda v: isinstance(v, (int, float)) and v > 0,
        "subtotal": lambda v: isinstance(v, (int, float)) and v > 0,
        "impuestos": lambda v: isinstance(v, (int, float)) and v >= 0,
        "impuesto": lambda v: isinstance(v, (int, float)) and v >= 0,
        "monto": lambda v: isinstance(v, (int, float)) and v > 0,
        "moneda": lambda v: bool(v and len(str(v)) == 3),
        "items": lambda v: isinstance(v, list) and len(v) > 0,
        "partes": lambda v: isinstance(v, list) and len(v) >= 2,
        "clausulas_clave": lambda v: isinstance(v, list) and len(v) > 0,
    }

    for field_name, value in data.items():
        present = value is not None and value != "" and value != []

        check = plausibility_checks.get(field_name)
        plausible = check(value) if check and present else present

        score = 0.0
        if present:
            score = 1.0 if plausible else 0.5

        field_scores.append(FieldScore(
            field_name=field_name,
            present=present,
            plausible=plausible,
            score=score,
        ))

    total_score = sum(fs.score for fs in field_scores) / max(len(field_scores), 1)
    return round(total_score, 2), field_scores


def print_confidence_report(data: dict, schema_type: str):
    """Imprime reporte detallado de confianza por campo."""
    total_score, field_scores = score_extraction(data, schema_type)
    print(f"\n  Confianza detallada: {total_score:.0%}")
    for fs in field_scores:
        status = "OK" if fs.plausible else ("parcial" if fs.present else "falta")
        print(f"    {fs.field_name:<20} {status:<10} {fs.score:.1f}")

Troubleshooting del Proyecto

Problema 1: PDF escaneado se detecta como digital

Síntoma: Un PDF escaneado tiene texto OCR invisible embebido (OCR layer) y PyMuPDF lo extrae, pero el texto es basura.

Solución: Agrega un quality check: calcula alpha_ratio = sum(c.isalpha() for c in text) / len(text). Si es menor a 0.5, re-clasifica como escaneado y usa Vision. Integra este check en select_and_extract como post-validación del texto de PyMuPDF.

Problema 2: Campos de factura con formatos inconsistentes

Síntoma: El LLM devuelve "total": "$1,234.56" en lugar de "total": 1234.56.

Solución: Agrega un pre-procesador entre el JSON del LLM y Pydantic que limpie valores monetarios: value.replace("$", "").replace(",", "") y luego float(). Aplícalo a campos numéricos antes de pasar a schema_class(**raw_json).

Problema 3: Documento largo excede timeout

Síntoma: Un contrato de 50 páginas toma más de 60 segundos y falla por timeout del cliente OpenAI.

Solución: Ajusta el timeout del cliente (OpenAI(timeout=120.0)) y reduce CHUNK_SIZE_PAGES = 3 para documentos muy largos.

Problema 4: Vision API falla con imágenes de baja resolución

Síntoma: Fotografías de documentos tomadas con poca luz devuelven texto parcial o incorrecto.

Solución: Pre-procesa con PIL antes de enviar: ImageEnhance.Contrast(img).enhance(1.5) y ImageEnhance.Sharpness(img).enhance(2.0). Guarda como JPEG quality=95 antes de codificar a Base64.

Problema 5: Schema auto-detectado incorrectamente

Síntoma: Un recibo se clasifica como factura porque contiene la palabra "factura simplificada".

Solución: Usa pesos diferenciados por keyword en lugar de conteo simple:

SCHEMA_WEIGHTS: dict[str, dict[str, float]] = {
    "invoice": {"factura": 2.0, "proveedor": 1.5, "iva": 1.0, "nº factura": 2.0},
    "receipt": {"ticket": 2.0, "recibo": 2.0, "cajero": 1.5, "cambio": 1.0},
    "contract": {"contrato": 2.0, "cláusula": 2.0, "vigencia": 1.5, "firmante": 1.5},
}

Checklist de Completitud

Pipeline core:

  • Acepta PDF e imágenes (JPG, PNG, WEBP)
  • Detecta PDF digital vs escaneado vs imagen
  • Extrae texto con PyMuPDF (digital), Vision (escaneado/imagen)
  • Tesseract disponible como método alternativo
  • Schemas Pydantic: InvoiceSchema, ReceiptSchema, ContractSchema
  • Schema registry con auto-detección por keywords
  • Extracción estructurada con LLM + validación Pydantic
  • Fallback parcial cuando la validación falla

Documentos largos:

  • Chunking por páginas (CHUNK_SIZE_PAGES configurable)
  • Procesamiento independiente por chunk
  • Merge de resultados (escalares: primer no-nulo, listas: concatenar)

Calidad y métricas:

  • ExtractionResult con todos los campos especificados
  • Costo estimado por extracción
  • Confianza basada en campos extraídos
  • Latencia medida
  • Errores capturados sin crashes

Extensiones:

  • Batch processing con tracking de progreso y presupuesto
  • Confidence scoring detallado por campo

Ejercicios

Ejercicio 1: Agregar Schema de Recibo Médico (Fácil)

Crea un MedicalReceiptSchema con campos: paciente, doctor, hospital, fecha, diagnostico, medicamentos (lista con nombre, dosis, cantidad), total. Regístralo en SCHEMA_REGISTRY con keywords apropiadas.

Ver solución 
class Medicamento(BaseModel):
    nombre: str = Field(default="", description="Nombre del medicamento")
    dosis: Optional[str] = Field(default=None, description="Dosis indicada")
    cantidad: int = Field(default=1, description="Cantidad recetada")
    precio: float = Field(default=0.0)


class MedicalReceiptSchema(BaseModel):
    paciente: Optional[str] = Field(default=None, description="Nombre del paciente")
    doctor: Optional[str] = Field(default=None, description="Nombre del médico")
    hospital: Optional[str] = Field(default=None, description="Centro médico")
    fecha: Optional[str] = Field(default=None, description="Fecha YYYY-MM-DD")
    diagnostico: Optional[str] = Field(default=None, description="Diagnóstico")
    medicamentos: list[Medicamento] = Field(default_factory=list)
    total: Optional[float] = Field(default=None)


SCHEMA_REGISTRY["medical_receipt"] = MedicalReceiptSchema
SCHEMA_KEYWORDS["medical_receipt"] = [
    "paciente", "doctor", "médico", "receta", "diagnóstico",
    "medicamento", "dosis", "hospital", "clínica",
]

result = extract_document("receta_medica.pdf", schema_type="medical_receipt")
print(json.dumps(result.data, indent=2, ensure_ascii=False))

Ejercicio 2: Capa de Validación Post-Extracción (Medio)

Crea una función validate_extraction(data, schema_type) que aplique reglas de negocio sobre los datos extraídos: total debe ser >= subtotal, fecha no puede ser futura, items deben sumar aprox. el subtotal (±10%). Retorna lista de warnings y un booleano is_valid.

Ver solución 
from datetime import date


def validate_extraction(data: dict, schema_type: str) -> tuple[bool, list[str]]:
    """Valida datos extraídos contra reglas de negocio."""
    warnings = []

    if schema_type in ("invoice", "receipt"):
        total = data.get("total")
        subtotal = data.get("subtotal")

        if total is not None and subtotal is not None:
            if total < subtotal:
                warnings.append(
                    f"Total ({total}) menor que subtotal ({subtotal})"
                )

        fecha_str = data.get("fecha")
        if fecha_str:
            try:
                doc_date = date.fromisoformat(fecha_str)
                if doc_date > date.today():
                    warnings.append(f"Fecha futura detectada: {fecha_str}")
            except ValueError:
                warnings.append(f"Fecha con formato inválido: {fecha_str}")

        items = data.get("items", [])
        if items and subtotal is not None:
            items_sum = sum(
                item.get("monto", 0) or item.get("precio", 0) * item.get("cantidad", 1)
                for item in items
            )
            if items_sum > 0 and abs(items_sum - subtotal) / subtotal > 0.10:
                warnings.append(
                    f"Items suman {items_sum:.2f}, subtotal es {subtotal:.2f} (diferencia >10%)"
                )

    if schema_type == "contract":
        partes = data.get("partes", [])
        if len(partes) < 2:
            warnings.append("Contrato con menos de 2 partes identificadas")

    is_valid = len(warnings) == 0
    return is_valid, warnings


result = extract_document("factura.pdf", schema_type="invoice")
if result.success:
    is_valid, validation_warnings = validate_extraction(result.data, result.schema_used)
    print(f"Validación: {'PASS' if is_valid else 'WARN'}")
    for w in validation_warnings:
        print(f"  - {w}")

Resumen

En este proyecto construiste un Document Extractor completo que:

  • Detecta tipo de documento (PDF digital, PDF escaneado, imagen) usando PyMuPDF para análisis de texto embebido
  • Extrae texto con el método óptimo: PyMuPDF para digitales (gratis), Tesseract para OCR local, Vision API para máxima calidad
  • Define schemas Pydantic para facturas, recibos y contratos con auto-detección por keywords
  • Extrae datos estructurados con LLM + validación Pydantic, con fallback parcial cuando la validación falla
  • Maneja documentos largos con chunking por páginas y merge inteligente de resultados
  • Reporta métricas de costo, confianza, método usado y latencia en cada extracción

Este extractor es la base del Document Analyzer del Módulo 8, donde integrarás RAG para Q&A sobre contenido extraído y TTS para resúmenes hablados.

Próximo módulo: Módulo 4 — Generación de Imágenes. Del análisis pasas a la creación: DALL-E, Stable Diffusion, y pipelines de generación controlada.

Recursos Adicionales

  1. PyMuPDF Documentation — Extracción de texto y renderizado de PDF
  2. Tesseract OCR — Motor OCR open-source
  3. Pydantic V2 Docs — Modelos, validación y JSON Schema
  4. OpenAI Vision Guide — Procesamiento de imágenes con GPT-4
  5. OpenAI Structured Outputs — JSON mode y response format
  6. pdf2image — Conversión de PDF a imágenes para OCR
← Anterior7. Troubleshooting DocumentosSiguiente →1. Introducción: Generación de Imágenes