2. Especificaciones Técnicas

Descripción

Antes de escribir una línea de código, necesitas un contrato claro: qué entra, qué sale, qué componentes participan, cómo se comunican, y cuánto cuesta. Esta cápsula define las especificaciones técnicas completas del Document Analyzer Multimodal. Es el equivalente a un documento de diseño técnico (TDD) que en un equipo de ingeniería se revisa antes de implementar.

Por qué importa: Construir sin especificaciones lleva a decisiones ad-hoc, refactors constantes y componentes que no encajan. Definir inputs, outputs, modelos de datos y endpoints antes de implementar te permite validar el diseño, estimar costos, y asegurar que todos los módulos hablan el mismo idioma.

Conexión con el módulo: Esta cápsula es la referencia para las cápsulas 03-08. Cada componente que construyas debe respetar los contratos definidos aquí. Si necesitas cambiar una especificación durante la implementación, vuelve aquí, actualiza el contrato, y propaga el cambio.

Inputs del Sistema

1. Documento (obligatorio)

El input principal es un archivo de documento. El sistema soporta dos categorías:

Restricciones técnicas

SUPPORTED_EXTENSIONS = {".pdf", ".png", ".jpg", ".jpeg", ".webp"}
MAX_FILE_SIZE_BYTES = 50 * 1024 * 1024  # 50 MB
MAX_PDF_PAGES = 50
MAX_IMAGE_DIMENSION = 4096  # píxeles por lado (recomendado)

Tipos de contenido dentro de PDFs

Un PDF puede tener diferentes tipos de páginas:

La detección de tipo por página es crítica: un PDF de 20 páginas puede tener 15 con texto y 5 escaneadas. El DocumentProcessor debe manejar ambos casos dentro del mismo documento.

2. Pregunta (opcional)

Para el módulo de Q&A:

QUESTION_MAX_LENGTH = 500  # caracteres
QUESTION_MIN_LENGTH = 5    # caracteres mínimos para una pregunta válida

Si no se envía pregunta, el sistema procesa el documento sin Q&A. Si se envía, el documento se indexa automáticamente para responder.

3. Opciones de procesamiento

Flags que controlan qué operaciones ejecutar:

Modelo Pydantic del request

from pydantic import BaseModel, Field
from typing import Optional
from enum import Enum


class TTSVoice(str, Enum):
    ALLOY = "alloy"
    ECHO = "echo"
    FABLE = "fable"
    ONYX = "onyx"
    NOVA = "nova"
    SHIMMER = "shimmer"


class AnalyzeRequest(BaseModel):
    question: Optional[str] = Field(
        None, min_length=5, max_length=500,
        description="Pregunta sobre el documento para Q&A"
    )
    extract_structured: bool = Field(
        True, description="Extraer datos estructurados del documento"
    )
    generate_summary: bool = Field(
        True, description="Generar resumen ejecutivo"
    )
    generate_audio_summary: bool = Field(
        False, description="Generar resumen en audio (TTS)"
    )
    index_for_qa: bool = Field(
        True, description="Indexar para Q&A posterior"
    )
    audio_voice: TTSVoice = Field(
        TTSVoice.NOVA, description="Voz para síntesis de audio"
    )

Outputs del Sistema

Respuesta estándar

Toda respuesta del endpoint /analyze sigue este formato:

class DocumentMetadata(BaseModel):
    doc_id: str
    filename: str
    file_type: str
    pages_processed: int
    document_type: Optional[str] = None
    indexed: bool = False
    latency_seconds: float
    estimated_cost_usd: float


class StructuredData(BaseModel):
    document_type: str
    fields: dict
    confidence: Optional[float] = None


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


class AnalyzeResponse(BaseModel):
    success: bool
    doc_id: str
    extracted_data: Optional[StructuredData] = None
    summary: Optional[str] = None
    qa_result: Optional[QAResult] = None
    audio_summary_url: Optional[str] = None
    metadata: DocumentMetadata
    errors: list[str] = []

Ejemplo de respuesta completa

{
    "success": true,
    "doc_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
    "extracted_data": {
        "document_type": "factura",
        "fields": {
            "fecha": "2025-03-15",
            "numero_factura": "FAC-2025-0042",
            "proveedor": "Tech Solutions S.A.",
            "subtotal": 1500.00,
            "iva": 240.00,
            "total": 1740.00,
            "items": [
                {"descripcion": "Licencia software", "cantidad": 1, "precio": 1200.00},
                {"descripcion": "Soporte técnico", "cantidad": 1, "precio": 300.00}
            ]
        },
        "confidence": 0.95
    },
    "summary": "Factura FAC-2025-0042 de Tech Solutions S.A. por $1,740.00 MXN. Incluye licencia de software ($1,200) y soporte técnico ($300). Subtotal $1,500 + IVA $240. Fecha: 15 de marzo 2025.",
    "qa_result": {
        "question": "¿Cuál es el total de la factura?",
        "answer": "El total de la factura es $1,740.00 MXN, que incluye un subtotal de $1,500.00 más IVA de $240.00.",
        "sources": ["Página 1: Total: $1,740.00", "Página 1: Subtotal: $1,500.00 | IVA: $240.00"],
        "confidence": 0.98
    },
    "audio_summary_url": "/audio/a1b2c3d4_resumen.mp3",
    "metadata": {
        "doc_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
        "filename": "factura_marzo.pdf",
        "file_type": "pdf",
        "pages_processed": 1,
        "document_type": "factura",
        "indexed": true,
        "latency_seconds": 8.45,
        "estimated_cost_usd": 0.045
    },
    "errors": []
}

Respuesta con errores parciales

El sistema no falla completamente si un componente tiene problemas. Reporta éxito parcial:

{
    "success": true,
    "doc_id": "...",
    "extracted_data": null,
    "summary": "Resumen del documento...",
    "qa_result": null,
    "audio_summary_url": null,
    "metadata": {
        "doc_id": "...",
        "filename": "manual_tecnico.pdf",
        "file_type": "pdf",
        "pages_processed": 15,
        "document_type": "manual",
        "indexed": true,
        "latency_seconds": 22.1,
        "estimated_cost_usd": 0.08
    },
    "errors": [
        "Extracción estructurada no disponible para tipo 'manual'",
        "TTS falló: resumen excede límite de caracteres"
    ]
}

Componentes del Sistema

Tabla de componentes

Interfaz de cada componente

Cada componente expone una interfaz mínima y predecible:

class DocumentProcessor:
    def process(self, file_path: str) -> ProcessedDocument: ...

class VisionAnalyzer:
    def classify(self, content: ProcessedDocument) -> str: ...
    def extract_structured(self, content: ProcessedDocument, doc_type: str) -> StructuredData: ...

class RAGModule:
    def index(self, doc_id: str, content: ProcessedDocument) -> None: ...
    def query(self, question: str, doc_id: str = None) -> QAResult: ...

class AudioModule:
    def generate_summary_audio(self, text: str, voice: str = "nova") -> str: ...

class DocumentSummarizer:
    def summarize(self, content: ProcessedDocument) -> str: ...

class CostTracker:
    def track(self, operation: str, model: str, tokens: int) -> None: ...
    def get_total(self) -> float: ...

Modelo de datos intermedio

Los componentes se comunican a través de un modelo compartido:

class PageContent(BaseModel):
    page_number: int
    content_type: str  # "text" | "image"
    text: Optional[str] = None
    image_base64: Optional[str] = None


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


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

API REST (FastAPI)

Endpoints

POST /analyze

Endpoint principal. Recibe documento y opciones, retorna análisis completo.

POST /analyze
Content-Type: multipart/form-data

Campos:
  - file: UploadFile (PDF o imagen, requerido)
  - question: str (opcional, max 500 chars)
  - extract_structured: bool (default: true)
  - generate_summary: bool (default: true)
  - generate_audio_summary: bool (default: false)
  - index_for_qa: bool (default: true)
  - audio_voice: str (default: "nova")

Respuesta: 200 OK → AnalyzeResponse
Errores:
  - 400: Archivo inválido (formato, tamaño)
  - 413: Archivo excede tamaño máximo
  - 422: Parámetros de validación incorrectos
  - 500: Error interno de procesamiento
  - 503: Servicio no disponible (API keys inválidas)

POST /ask

Para hacer preguntas sobre documentos ya indexados:

POST /ask
Content-Type: application/json

Body:
{
    "question": "¿Cuál es la fecha de vencimiento?",
    "doc_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890"  // opcional
}

Respuesta: 200 OK → QAResult
Errores:
  - 400: Pregunta vacía o inválida
  - 404: doc_id no encontrado (si se especificó)
  - 500: Error en generación de respuesta

GET /health

GET /health

Respuesta: 200 OK
{
    "status": "ok",
    "openai": "connected",
    "chromadb": "connected",
    "indexed_documents": 5,
    "uptime_seconds": 3600
}

Respuesta degradada: 503
{
    "status": "degraded",
    "openai": "error: invalid API key",
    "chromadb": "connected"
}

Diagrama de Arquitectura

Flujo completo de un request POST /analyze

Cliente
  │
  ├─── POST /analyze (file=factura.pdf, question="¿Total?", extract=true, audio=true)
  │
  ▼
┌─────────────────────────────────────────────────────────────┐
│                    FastAPI Router                            │
│  1. Validar archivo (tipo, tamaño)                         │
│  2. Guardar archivo temporal                                │
│  3. Generar doc_id                                         │
│  4. Iniciar CostTracker                                    │
└──────────┬──────────────────────────────────────────────────┘
           │
           ▼
┌─────────────────────────────────────────────────────────────┐
│              DocumentProcessor.process()                     │
│  - Detectar tipo (PDF vs imagen)                            │
│  - Para PDF: iterar páginas                                 │
│    - Página con texto → extraer texto                       │
│    - Página escaneada → convertir a imagen base64           │
│  - Para imagen: codificar a base64                          │
│  - Retornar ProcessedDocument                               │
└──────────┬──────────────────────────────────────────────────┘
           │
     ┌─────┴─────────────────────────────────────┐
     │                                           │
     ▼                                           ▼
┌──────────────────────┐              ┌──────────────────────┐
│ VisionAnalyzer       │              │ RAGModule            │
│ .classify() → tipo   │              │ .index() → ChromaDB  │
│ .extract_structured()│              │                      │
│  → StructuredData    │              │ Si hay pregunta:     │
└──────────┬───────────┘              │ .query() → QAResult  │
           │                          └──────────┬───────────┘
           ▼                                     │
┌──────────────────────┐                         │
│ DocumentSummarizer   │                         │
│ .summarize()         │                         │
│  → resumen texto     │                         │
└──────────┬───────────┘                         │
           │                                     │
           ▼                                     │
┌──────────────────────┐                         │
│ AudioModule          │                         │
│ .generate_summary_   │                         │
│  audio()             │                         │
│  → archivo .mp3      │                         │
└──────────┬───────────┘                         │
           │                                     │
           └─────────────┬───────────────────────┘
                         │
                         ▼
              ┌──────────────────────┐
              │  Construir           │
              │  AnalyzeResponse     │
              │  + metadata          │
              │  + costos            │
              └──────────┬───────────┘
                         │
                         ▼
                     200 OK → JSON

Stack Tecnológico

Dependencias principales

Dependencias opcionales

Modelos de IA utilizados

Presupuesto de Costos

Por operación

Escenarios de costo

Límite mensual recomendado

Para un servicio en producción con ~1000 documentos/mes:

1000 docs × $0.05 promedio = $50/mes
+ Buffer 20% por retries     = $60/mes

Schemas de Extracción por Tipo de Documento

Factura

INVOICE_SCHEMA = {
    "fecha": "str (YYYY-MM-DD)",
    "numero_factura": "str",
    "proveedor": "str",
    "receptor": "str",
    "subtotal": "float",
    "impuestos": "float",
    "total": "float",
    "moneda": "str (MXN, USD, EUR)",
    "items": [
        {
            "descripcion": "str",
            "cantidad": "int",
            "precio_unitario": "float",
            "total_linea": "float"
        }
    ]
}

Contrato

CONTRACT_SCHEMA = {
    "tipo_contrato": "str",
    "partes": ["str"],
    "fecha_firma": "str (YYYY-MM-DD)",
    "fecha_vigencia": "str (YYYY-MM-DD)",
    "objeto": "str",
    "monto": "float o null",
    "clausulas_clave": ["str"]
}

Manual / Informe

REPORT_SCHEMA = {
    "titulo": "str",
    "autor": "str o null",
    "fecha": "str (YYYY-MM-DD) o null",
    "secciones": ["str"],
    "resumen_ejecutivo": "str"
}

Registro de schemas

EXTRACTION_SCHEMAS: dict[str, dict] = {
    "factura": INVOICE_SCHEMA,
    "contrato": CONTRACT_SCHEMA,
    "manual": REPORT_SCHEMA,
    "informe": REPORT_SCHEMA,
}

def get_schema_for_type(doc_type: str) -> dict:
    return EXTRACTION_SCHEMAS.get(
        doc_type.lower(),
        {"contenido_general": "str", "puntos_clave": ["str"]}
    )

Ejercicios

Ejercicio 1: Validador de input completo

Implementa una función que valide el archivo de input verificando: existencia, extensión soportada, tamaño máximo, y para PDFs, número máximo de páginas. Debe retornar una lista de errores (vacía si todo es válido).

import fitz
from pathlib import Path


def validate_input(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. "
            f"Formatos válidos: {', '.join(SUPPORTED_EXTENSIONS)}"
        )
        return errors

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

    if file_size == 0:
        errors.append("Archivo vacío")
        return errors

    if p.suffix.lower() == ".pdf":
        try:
            doc = fitz.open(file_path)
            if len(doc) > MAX_PDF_PAGES:
                errors.append(
                    f"PDF de {len(doc)} páginas excede "
                    f"el límite de {MAX_PDF_PAGES} páginas"
                )
            if len(doc) == 0:
                errors.append("PDF sin páginas")
            doc.close()
        except Exception as e:
            errors.append(f"PDF corrupto o ilegible: {e}")

    return errors

Ejercicio 2: Modelo Pydantic completo para la respuesta

Define todos los modelos Pydantic necesarios para la respuesta del endpoint /analyze, incluyendo validaciones custom: doc_id debe ser UUID válido, latency_seconds debe ser positivo, estimated_cost_usd no negativo.

from pydantic import BaseModel, Field, field_validator
from typing import Optional
import uuid


class DocumentMetadata(BaseModel):
    doc_id: str
    filename: str
    file_type: str
    pages_processed: int = Field(ge=0)
    document_type: Optional[str] = None
    indexed: bool = False
    latency_seconds: float = Field(ge=0)
    estimated_cost_usd: float = Field(ge=0)

    @field_validator("doc_id")
    @classmethod
    def validate_uuid(cls, v: str) -> str:
        uuid.UUID(v)
        return v

    @field_validator("file_type")
    @classmethod
    def validate_file_type(cls, v: str) -> str:
        if v not in ("pdf", "png", "jpg", "jpeg", "webp"):
            raise ValueError(f"Tipo de archivo inválido: {v}")
        return v


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


class QAResult(BaseModel):
    question: str = Field(min_length=1)
    answer: str
    sources: list[str] = []
    confidence: Optional[float] = Field(None, ge=0, le=1)


class AnalyzeResponse(BaseModel):
    success: bool
    doc_id: str
    extracted_data: Optional[StructuredData] = None
    summary: Optional[str] = None
    qa_result: Optional[QAResult] = None
    audio_summary_url: Optional[str] = None
    metadata: DocumentMetadata
    errors: list[str] = []

Resumen

• El Document Analyzer acepta PDFs (hasta 50 páginas) e imágenes (hasta 20 MB) como input.
• La respuesta incluye: datos estructurados, resumen, Q&A con fuentes, URL de audio, y metadata con costos.
• 6 componentes con interfaces definidas: DocumentProcessor, VisionAnalyzer, RAGModule, AudioModule, Summarizer, CostTracker.
• 5 endpoints REST: /analyze (principal), /ask (Q&A), /documents/{id}, /health, /audio/{filename}.
• Los schemas de extracción varían por tipo de documento: factura, contrato, manual/informe.
• Costo promedio: $0.03-0.08 por documento. ~$50-60/mes para 1000 documentos.
• Todos los modelos Pydantic están definidos para validación estricta de inputs y outputs.

Recursos Adicionales

1. FastAPI Request Files — Upload de archivos
2. Pydantic V2 — Validación de modelos
3. OpenAI Pricing — Costos actualizados de modelos
4. PyMuPDF — Procesamiento de PDFs