4. Análisis con Vision

Descripción

El VisionAnalyzer es el componente inteligente del Document Analyzer: recibe las imágenes extraídas por el DocumentProcessor y las transforma en información útil. Clasifica el tipo de documento (factura, contrato, manual), extrae datos estructurados según el tipo, y genera descripciones de imágenes para indexar en RAG. Usa GPT-4o como proveedor principal con fallback a Claude y Gemini.

Por qué importa: Un PDF escaneado sin Vision es solo una colección de píxeles. El VisionAnalyzer le da significado: "esto es una factura de $1,740 de Tech Solutions". Sin este componente, los documentos escaneados serían inútiles para el sistema. Y con fallback multi-proveedor, el sistema sigue funcionando aunque un proveedor falle.

Conexión con el módulo: En el Módulo 2 aprendiste a enviar imágenes a GPT-4 Vision y obtener análisis. Aquí integras esa capacidad en una clase con tres funcionalidades: clasificación, extracción estructurada, y descripción para RAG. Además, implementas el patrón de fallback del Módulo 7: si OpenAI falla o está lento, el sistema automáticamente intenta con Anthropic o Google.

Arquitectura del VisionAnalyzer

Responsabilidades

VisionAnalyzer
├── classify()              → Determinar tipo de documento
├── extract_structured()    → Extraer datos según tipo (factura, contrato, etc.)
├── describe_for_rag()      → Generar descripciones textuales para indexación
└── _call_with_fallback()   → Ejecutar con fallback multi-proveedor

Flujo de decisión

ProcessedDocument
    │
    ├── ¿Tiene imágenes? ──── Sí ──→ classify() con primera imagen
    │                                    │
    │                                    ▼
    │                          extract_structured() con schema del tipo
    │                                    │
    │                                    ▼
    │                          describe_for_rag() para cada imagen
    │
    └── ¿Solo texto? ──────── Sí ──→ classify() con texto (sin Vision)
                                         │
                                         ▼
                               extract_structured() con LLM texto

Modelos Pydantic para Extracción

Modelos base

from pydantic import BaseModel, Field
from typing import Optional


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


class InvoiceData(BaseModel):
    fecha: Optional[str] = None
    numero_factura: Optional[str] = None
    proveedor: Optional[str] = None
    receptor: Optional[str] = None
    subtotal: Optional[float] = None
    impuestos: Optional[float] = None
    total: Optional[float] = None
    moneda: str = "MXN"
    items: list[dict] = []


class ContractData(BaseModel):
    tipo_contrato: Optional[str] = None
    partes: list[str] = []
    fecha_firma: Optional[str] = None
    fecha_vigencia: Optional[str] = None
    objeto: Optional[str] = None
    monto: Optional[float] = None
    clausulas_clave: list[str] = []


class ReportData(BaseModel):
    titulo: Optional[str] = None
    autor: Optional[str] = None
    fecha: Optional[str] = None
    secciones: list[str] = []
    resumen_ejecutivo: Optional[str] = None

Registro de schemas

EXTRACTION_SCHEMAS: dict[str, type[BaseModel]] = {
    "factura": InvoiceData,
    "contrato": ContractData,
    "manual": ReportData,
    "informe": ReportData,
}

SCHEMA_PROMPTS: dict[str, str] = {
    "factura": "fecha, numero_factura, proveedor, receptor, subtotal, impuestos, total, moneda, items (descripcion, cantidad, precio_unitario, total_linea)",
    "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",
}

DEFAULT_SCHEMA_PROMPT = "titulo, contenido_principal, puntos_clave, fecha (si aparece)"

Implementación: VisionAnalyzer

Clase completa

import json
import logging
import os
from typing import Optional

from openai import OpenAI

logger = logging.getLogger(__name__)


class VisionAnalyzer:
    def __init__(self):
        self.openai_client = OpenAI()
        self.anthropic_client = None
        self.google_model = None
        self._init_fallback_providers()

    def _init_fallback_providers(self):
        try:
            import anthropic
            if os.getenv("ANTHROPIC_API_KEY"):
                self.anthropic_client = anthropic.Anthropic()
                logger.info("Anthropic disponible como fallback")
        except ImportError:
            logger.info("Anthropic no instalado — sin fallback")

        try:
            import google.generativeai as genai
            if os.getenv("GOOGLE_API_KEY"):
                genai.configure(api_key=os.getenv("GOOGLE_API_KEY"))
                self.google_model = genai.GenerativeModel("gemini-1.5-flash")
                logger.info("Google Gemini disponible como fallback")
        except ImportError:
            logger.info("Google GenAI no instalado — sin fallback")

    def classify(self, content) -> str:
        if content.has_image_pages:
            images = content.get_images_for_vision()
            return self._classify_from_image(images[0]["base64"])

        if content.full_text:
            return self._classify_from_text(content.full_text[:2000])

        return "otro"

    def _classify_from_image(self, image_base64: str) -> str:
        prompt = (
            "Clasifica este documento en una de estas categorías: "
            "factura, contrato, manual, informe, otro. "
            "Responde SOLO con el nombre de la categoría, sin explicación."
        )

        def openai_call():
            r = self.openai_client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[{
                    "role": "user",
                    "content": [
                        {"type": "text", "text": prompt},
                        {"type": "image_url", "image_url": {
                            "url": f"data:image/png;base64,{image_base64}"
                        }}
                    ]
                }],
                max_tokens=20,
                temperature=0
            )
            return r.choices[0].message.content.strip().lower()

        def anthropic_call():
            if not self.anthropic_client:
                raise RuntimeError("Anthropic no disponible")
            r = self.anthropic_client.messages.create(
                model="claude-3-5-sonnet-20241022",
                max_tokens=20,
                messages=[{
                    "role": "user",
                    "content": [
                        {"type": "image", "source": {
                            "type": "base64", "media_type": "image/png",
                            "data": image_base64
                        }},
                        {"type": "text", "text": prompt}
                    ]
                }]
            )
            return r.content[0].text.strip().lower()

        return self._call_with_fallback([openai_call, anthropic_call], "clasificación")

    def _classify_from_text(self, text: str) -> str:
        r = self.openai_client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{
                "role": "user",
                "content": (
                    "Clasifica este documento: factura, contrato, manual, informe, otro. "
                    f"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_prompt = SCHEMA_PROMPTS.get(doc_type, DEFAULT_SCHEMA_PROMPT)

        if content.has_image_pages:
            images = content.get_images_for_vision()
            data = self._extract_from_images(images[:5], schema_prompt)
        elif content.full_text:
            data = self._extract_from_text(content.full_text[:4000], schema_prompt)
        else:
            return ExtractionResult(
                document_type=doc_type, fields={},
                confidence=0, raw_response="Sin contenido para extraer"
            )

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

    def _extract_from_images(self, images: list[dict], schema_prompt: str) -> dict:
        prompt = (
            f"Extrae los siguientes campos de este documento: {schema_prompt}\n\n"
            "Responde ÚNICAMENTE con JSON válido. "
            "Usa null para campos no encontrados. "
            "Para listas vacías usa []."
        )

        content_parts = [{"type": "text", "text": prompt}]
        for img in images:
            content_parts.append({
                "type": "image_url",
                "image_url": {"url": f"data:image/png;base64,{img['base64']}"}
            })

        def openai_call():
            r = self.openai_client.chat.completions.create(
                model="gpt-4o",
                messages=[{"role": "user", "content": content_parts}],
                response_format={"type": "json_object"},
                temperature=0,
                max_tokens=2000
            )
            return json.loads(r.choices[0].message.content)

        def anthropic_call():
            if not self.anthropic_client:
                raise RuntimeError("Anthropic no disponible")
            ant_content = []
            for img in images:
                ant_content.append({
                    "type": "image", "source": {
                        "type": "base64", "media_type": "image/png",
                        "data": img["base64"]
                    }
                })
            ant_content.append({"type": "text", "text": prompt + "\nResponde solo JSON."})

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

        return self._call_with_fallback([openai_call, anthropic_call], "extracción estructurada")

    def _extract_from_text(self, text: str, schema_prompt: str) -> dict:
        r = self.openai_client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{
                "role": "user",
                "content": (
                    f"Extrae los siguientes campos: {schema_prompt}\n\n"
                    "Responde ÚNICAMENTE con JSON válido. Usa null para no encontrados.\n\n"
                    f"{text}"
                )
            }],
            response_format={"type": "json_object"},
            temperature=0,
            max_tokens=2000
        )
        return json.loads(r.choices[0].message.content)

    def describe_for_rag(self, images: list[dict]) -> list[str]:
        descriptions = []
        for img in images[:10]:
            try:
                desc = self._describe_single_image(img["base64"])
                descriptions.append(f"[Página {img.get('page', '?')}] {desc}")
            except Exception as e:
                logger.warning(f"Error describiendo imagen página {img.get('page', '?')}: {e}")
                descriptions.append(f"[Página {img.get('page', '?')}] Imagen no descrita por error.")
        return descriptions

    def _describe_single_image(self, image_base64: str) -> str:
        r = self.openai_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. "
                        "Incluye: tipo de documento, datos visibles, estructura."
                    )},
                    {"type": "image_url", "image_url": {
                        "url": f"data:image/png;base64,{image_base64}"
                    }}
                ]
            }],
            max_tokens=200,
            temperature=0
        )
        return r.choices[0].message.content

    def _estimate_confidence(self, data: dict, doc_type: str) -> float:
        if not data:
            return 0.0

        expected_fields = {
            "factura": ["fecha", "total", "proveedor"],
            "contrato": ["partes", "fecha_firma", "objeto"],
            "manual": ["titulo", "secciones"],
            "informe": ["titulo", "secciones"],
        }

        required = expected_fields.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)

    def _call_with_fallback(self, providers: list, operation: str):
        last_error = None
        for i, provider_fn in enumerate(providers):
            try:
                result = provider_fn()
                if i > 0:
                    logger.info(f"{operation}: éxito con proveedor fallback #{i}")
                return result
            except Exception as e:
                last_error = e
                logger.warning(f"{operation}: proveedor #{i} falló: {e}")
                continue

        raise RuntimeError(
            f"{operation}: todos los proveedores fallaron. Último error: {last_error}"
        )

Extracción Multi-Página

El problema

Un documento de 10 páginas puede tener información distribuida: la fecha está en la página 1, los items en las páginas 2-8, y el total en la página 9. Enviar solo la primera imagen pierde información.

Estrategia

Para documentos con múltiples imágenes, enviamos hasta 5 imágenes en una sola llamada a Vision. Si tiene más de 5, priorizamos:

1. Primera página (encabezado, datos generales)
2. Última página (totales, firmas)
3. Páginas intermedias seleccionadas

def _select_pages_for_extraction(self, images: list[dict], max_pages: int = 5) -> list[dict]:
    if len(images) <= max_pages:
        return images

    selected = [images[0], images[-1]]

    remaining = images[1:-1]
    step = max(1, len(remaining) // (max_pages - 2))
    for i in range(0, len(remaining), step):
        if len(selected) >= max_pages:
            break
        selected.append(remaining[i])

    selected.sort(key=lambda x: x.get("page", 0))
    return selected

Costo de multi-página

Fallback Multi-Proveedor en Detalle

Por qué necesitas fallback

Orden de preferencia

1. OpenAI GPT-4o       → Mejor calidad general, más caro
2. Anthropic Claude 3.5 → Comparable calidad, buen fallback
3. Google Gemini Flash  → Más barato, bueno para clasificación

Configuración del fallback

PROVIDER_CONFIG = {
    "openai": {
        "timeout": 30,
        "max_retries": 1,
        "models": {
            "vision": "gpt-4o",
            "classify": "gpt-4o-mini",
            "describe": "gpt-4o-mini"
        }
    },
    "anthropic": {
        "timeout": 30,
        "max_retries": 1,
        "models": {
            "vision": "claude-3-5-sonnet-20241022",
            "classify": "claude-3-5-sonnet-20241022"
        }
    },
    "google": {
        "timeout": 20,
        "max_retries": 1,
        "models": {
            "vision": "gemini-1.5-flash",
            "classify": "gemini-1.5-flash"
        }
    }
}

Troubleshooting

"La extracción retorna campos vacíos o null"

Causa probable: La imagen es de baja resolución o el documento tiene texto pequeño.

Solución: Aumentar la resolución de renderizado:

processor = DocumentProcessor(dpi_scale=300/72)  # 300 DPI en vez de 150

O usar el modelo más capaz:

# Cambiar de gpt-4o-mini a gpt-4o para extracción
r = self.openai_client.chat.completions.create(
    model="gpt-4o",  # más capaz para documentos complejos
    ...
)

"El clasificador retorna 'otro' para documentos comunes"

Causa probable: El prompt de clasificación necesita más contexto.

Solución: Mejorar el prompt con ejemplos:

CLASSIFICATION_PROMPT = """Clasifica este documento en una categoría:
- factura: documentos de cobro con montos, items, IVA
- contrato: acuerdos legales entre partes con cláusulas
- manual: documentación técnica con instrucciones
- informe: reportes con datos, gráficos, conclusiones
- otro: cualquier documento que no encaje en las anteriores

Responde SOLO con el nombre de la categoría."""

"Fallback a Anthropic falla con error de formato de imagen"

Causa probable: La imagen es JPEG pero se envía como image/png.

Solución: Detectar el formato real:

import imghdr

def detect_media_type(image_base64: str) -> str:
    raw = base64.b64decode(image_base64[:100])
    img_type = imghdr.what(None, h=raw)
    media_types = {
        "jpeg": "image/jpeg",
        "png": "image/png",
        "webp": "image/webp",
        "gif": "image/gif"
    }
    return media_types.get(img_type, "image/png")

"Rate limit (429) en extracción de múltiples documentos"

Solución: Agregar rate limiting del lado del cliente:

import time

def extract_batch(self, documents: list, delay: float = 1.0) -> list[ExtractionResult]:
    results = []
    for doc in documents:
        result = self.extract_structured(doc, doc.get("type", "otro"))
        results.append(result)
        time.sleep(delay)
    return results

Uso del VisionAnalyzer

Ejemplo completo

processor = DocumentProcessor()
analyzer = VisionAnalyzer()

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

doc_type = analyzer.classify(doc)
print(f"Tipo de documento: {doc_type}")

extraction = analyzer.extract_structured(doc, doc_type)
print(f"Confianza: {extraction.confidence}")
print(f"Datos extraídos:")
for key, value in extraction.fields.items():
    print(f"  {key}: {value}")

if doc.has_image_pages:
    descriptions = analyzer.describe_for_rag(doc.get_images_for_vision())
    print(f"\nDescripciones para RAG:")
    for desc in descriptions:
        print(f"  {desc}")

Output esperado

Tipo de documento: factura
Confianza: 1.0
Datos extraídos:
  fecha: 2025-03-15
  numero_factura: FAC-2025-0042
  proveedor: Tech Solutions S.A.
  receptor: Empresa ABC
  subtotal: 1500.0
  impuestos: 240.0
  total: 1740.0
  moneda: MXN
  items: [{'descripcion': 'Licencia software', 'cantidad': 1, ...}]

Descripciones para RAG:
  [Página 1] Factura comercial de Tech Solutions S.A. con número FAC-2025-0042...

Ejercicios

Ejercicio 1: Schema dinámico por clasificación

Implementa un flujo completo que: (1) clasifica el documento, (2) selecciona el schema correcto, (3) extrae datos, (4) valida con el modelo Pydantic correspondiente. Si la clasificación es desconocida, usa un schema genérico.

def analyze_document(content, analyzer: VisionAnalyzer) -> ExtractionResult:
    doc_type = analyzer.classify(content)
    logger.info(f"Documento clasificado como: {doc_type}")

    extraction = analyzer.extract_structured(content, doc_type)

    model_class = EXTRACTION_SCHEMAS.get(doc_type)
    if model_class:
        try:
            validated = model_class(**extraction.fields)
            extraction.fields = validated.model_dump()
            logger.info(f"Datos validados con {model_class.__name__}")
        except Exception as e:
            logger.warning(f"Validación falló: {e}. Usando datos crudos.")
    else:
        logger.info(f"Sin schema específico para '{doc_type}', datos sin validación extra")

    return extraction


processor = DocumentProcessor()
analyzer = VisionAnalyzer()

doc = processor.process("documento.pdf")
result = analyze_document(doc, analyzer)
print(f"Tipo: {result.document_type}")
print(f"Campos: {json.dumps(result.fields, indent=2, ensure_ascii=False)}")
print(f"Confianza: {result.confidence}")

Ejercicio 2: Fallback completo con Google Gemini

Extiende el _extract_from_images para incluir Google Gemini como tercer proveedor de fallback. Gemini usa una API diferente: acepta imágenes como PIL.Image o como bytes con upload_file. Implementa la función de Gemini y agrégala a la cadena de fallback.

import base64
from PIL import Image
import io


def gemini_extract(self, images: list[dict], schema_prompt: str) -> dict:
    if not self.google_model:
        raise RuntimeError("Google Gemini no disponible")

    pil_images = []
    for img in images[:5]:
        raw = base64.b64decode(img["base64"])
        pil_images.append(Image.open(io.BytesIO(raw)))

    prompt = (
        f"Extrae los siguientes campos de este documento: {schema_prompt}\n"
        "Responde ÚNICAMENTE con JSON válido. Usa null para no encontrados."
    )

    content_parts = pil_images + [prompt]

    response = self.google_model.generate_content(
        content_parts,
        generation_config={"temperature": 0, "max_output_tokens": 2000}
    )

    text = response.text
    if text.startswith("```"):
        text = text.split("\n", 1)[1].rsplit("```", 1)[0]

    return json.loads(text)


def _extract_from_images_with_gemini(self, images, schema_prompt):
    def openai_call():
        return self._extract_from_images_openai(images, schema_prompt)

    def anthropic_call():
        return self._extract_from_images_anthropic(images, schema_prompt)

    def google_call():
        return gemini_extract(self, images, schema_prompt)

    return self._call_with_fallback(
        [openai_call, anthropic_call, google_call],
        "extracción estructurada"
    )

Resumen

• El VisionAnalyzer transforma imágenes de documentos en datos estructurados.
• Tres funciones principales: clasificar tipo, extraer datos según schema, describir para RAG.
• Fallback multi-proveedor: OpenAI → Anthropic → Google, automático y transparente.
• Los schemas de extracción varían por tipo de documento (factura, contrato, manual).
• La extracción multi-página selecciona las páginas más informativas para optimizar costos.
• La confianza se estima comparando campos encontrados vs campos esperados.
• Troubleshooting: resolución de imagen, prompts de clasificación, formato de imagen, rate limits.

Recursos Adicionales

1. OpenAI Vision Guide — GPT-4 Vision
2. Anthropic Vision Docs — Claude Vision
3. Google Gemini Vision — Gemini multimodal
4. Módulo 2 de esta guía — Base de Vision