PDF LLM Extraction

PDF → Structured Data via LLM

Resumen

Pipeline para extraer datos estructurados de PDFs digitales usando análisis de fuentes + LLM como motor principal. Reemplaza regex/heurísticas por un enfoque que funciona con cualquier formato de PDF sin configuración manual por tipo.

Validado: 270 informes CIAF (2007-2025) → 99.6% éxito (231/232), ~0.3s extracción + ~15-20s LLM por PDF. Batch processing completo en ~71 minutos. Texto ampliado a 60K chars (antes 28K) para capturar conclusiones y recomendaciones completas.

Cuándo usarlo

• Usuario tiene PDFs con texto seleccionable (no escaneados) y quiere datos estructurados (JSON, CSV)
• Usuario dice "los PDF no me dan bien los datos" o "regex no funciona" → este skill es la respuesta
• Pipeline de extracción masiva (cientos/miles de PDFs del mismo tipo)
• Auto-learn de schema: el sistema detecta la estructura del documento sin configuración previa

No es para

• PDFs escaneados (imágenes) → usar ocr-quirurgico-pdf-md
• Conversión PDF → Markdown navegable → usar ocr-quirurgico-pdf-md
• Generación de artefactos de contenido (LinkedIn, infografías) → usar pdf-to-artifacts-david-antizar
• Extracción rápida sin estructura → usar markitdown

El Problema que Resuelve

Antes (regex):

# ❌ Frágil: rompe con cualquier variación de formato
re.search(r'Fecha:\s*(\d{2}/\d{2}/\d{4})', text)
re.search(r'N.*?informe.*?IF-(\d+)', text)

• ~55-72% de extracción exitosa en informes CIAF
• Requiere ajuste manual por tipo de PDF
• No maneja variaciones de formato

Ahora (LLM):

# ✅ Robusto: el LLM entiende contexto y variaciones
prompt = f"Extrae los campos del schema del siguiente texto:\n{text}"
response = llm_call(prompt, schema=CIAF_SCHEMA)

• 100% de confianza en informes CIAF 2024
• Funciona con cualquier variación de formato
• Auto-learn: detecta schema del primer lote de PDFs

Arquitectura del Pipeline

PDF digital
    ↓
┌─────────────────────────────────┐
│  FASE 1 — EXTRACCIÓN PURA       │
│  PyMuPDF: texto + font metadata  │
│  • page.get_text("dict")         │
│  • spans: text, font, size, bold │
│  • ~0.3s por PDF                 │
└─────────────────────────────────┘
    ↓
┌─────────────────────────────────┐
│  FASE 2 — ANÁLISIS ESTRUCTURAL  │
│  Font size clustering:           │
│  • Mediana de font sizes         │
│  • Umbral = mediana × 1.3       │
│  • textos mayores → headings     │
│  • Detecta secciones auto        │
│  • ~0.05s por PDF                │
└─────────────────────────────────┘
    ↓
┌─────────────────────────────────┐
│  FASE 3 — CHUNKING INTELIGENTE  │
│  Agrupar líneas por sección:     │
│  • Cada heading inicia chunk     │
│  • Chunks ≤ 6000 chars (para LLM)│
│  • Priorizar secciones relevantes│
└─────────────────────────────────┘
    ↓
┌─────────────────────────────────┐
│  FASE 4 — LLM STRUCTURING       │
│  Qwen 3.6 via NaN API:          │
│  • Prompt con schema + texto     │
│  • Respuesta JSON directa        │
│  • ~8-15s por PDF                │
│  • Modelo: "qwen3.6"             │
│  • API: api.nan.builders/v1      │
└─────────────────────────────────┘
    ↓
┌─────────────────────────────────┐
│  FASE 5 — VALIDACIÓN            │
│  JSON Schema validation:         │
│  • Campos requeridos presentes   │
│  • Tipos correctos (ISO dates)   │
│  • Arrays no vacíos              │
│  • Cross-field consistency       │
└─────────────────────────────────┘
    ↓
┌─────────────────────────────────┐
│  FASE 6 — EXPORT                │
│  • JSON (array de resultados)    │
│  • CSV (una fila por informe)    │
│  • Reporte de confianza          │
└─────────────────────────────────┘

Fase 0 — Análisis de longitudes de texto (OBLIGATORIO antes del batch)

ANTES de procesar, SIEMPRE analizar las longitudes de texto del corpus:

import fitz, os, statistics

pdf_dir = "/ruta/a/pdfs"
lengths = []

for fname in os.listdir(pdf_dir):
    if not fname.endswith(".pdf"):
        continue
    doc = fitz.open(os.path.join(pdf_dir, fname))
    text = ""
    for page in doc:
        text += page.get_text()
    doc.close()
    lengths.append(len(text))

print(f"Mínimo: {min(lengths):,} chars")
print(f"Máximo: {max(lengths):,} chars")
print(f"Promedio: {statistics.mean(lengths):,.0f} chars")
print(f"Mediana: {statistics.median(lengths):,.0f} chars")
p95 = sorted(lengths)[int(len(lengths)*0.95)]
print(f"P95: {p95:,} chars")
print(f"→ Límite recomendado: {p95 + 5000:,} chars (P95 + margen)")

Resultados CIAF (270 PDFs): Mínimo 13K, Máximo 266K, Mediana 48K, P95 143K → Límite: 60K chars.

Regla: Si el P95 > 50K, usar 60K como límite. Si P95 < 30K, usar 30K. El límite debe cubrir la mayoría de los documentos sin truncar las secciones finales (conclusiones, recomendaciones).

Fase 1 — Extracción de texto + font metadata

import fitz
import json

TEXT_LIMIT = 60000  # chars — ajustar según análisis de Fase 0

def extract_text_and_fonts(pdf_path: str) -> dict:
    """Extrae texto plano + metadata de fuentes de cada página."""
    doc = fitz.open(pdf_path)
    pages = []
    
    for page_num, page in enumerate(doc):
        text_dict = page.get_text("dict")
        lines = []
        
        for block in text_dict.get("blocks", []):
            if block.get("type") != 0:  # solo texto
                continue
            for line in block.get("lines", []):
                line_text = ""
                line_fonts = []
                for span in line.get("spans", []):
                    line_text += span.get("text", "")
                    line_fonts.append({
                        "text": span.get("text", ""),
                        "font": span.get("font", ""),
                        "size": round(span.get("size", 12), 1),
                        "bold": "Bold" in span.get("font", "") or "bold" in span.get("font", "").lower(),
                        "color": span.get("color", 0)
                    })
                if line_text.strip():
                    lines.append({
                        "text": line_text.strip(),
                        "max_font_size": max(f["size"] for f in line_fonts) if line_fonts else 12,
                        "is_bold": any(f["bold"] for f in line_fonts),
                        "fonts": line_fonts
                    })
        
        pages.append({"page": page_num + 1, "lines": lines})
    
    doc.close()
    return {"pages": pages, "total_pages": len(pages)}

Fase 2 — Detección de secciones por font size

def detect_sections(pages_data: dict) -> list:
    """
    Detecta headings usando font size clustering.
    
    Lógica:
    1. Recoger todos los font sizes del documento
    2. Calcular mediana (el "body text size")
    3. Cualquier texto con size > mediana × 1.3 → heading
    4. Agrupar headings + body text en secciones
    """
    all_sizes = []
    for page in pages_data["pages"]:
        for line in page["lines"]:
            all_sizes.append(line["max_font_size"])
    
    if not all_sizes:
        return []
    
    median_size = sorted(all_sizes)[len(all_sizes) // 2]
    heading_threshold = median_size * 1.3
    
    sections = []
    current_section = {"title": " preamble", "lines": []}
    
    for page in pages_data["pages"]:
        for line in page["lines"]:
            if line["max_font_size"] >= heading_threshold:
                # Guardar sección anterior
                if current_section["lines"]:
                    sections.append(current_section)
                current_section = {
                    "title": line["text"],
                    "lines": [],
                    "is_bold": line["is_bold"],
                    "page": page["page"]
                }
            else:
                current_section["lines"].append(line["text"])
    
    if current_section["lines"]:
        sections.append(current_section)
    
    return sections

Fase 3 — LLM Structuring

import requests
import os
import json

NAN_API = os.getenv("NAN_API")

def llm_extract_structured(text: str, schema: dict, model: str = "qwen3.6") -> dict:
    """
    Envía texto + schema al LLM y devuelve datos estructurados.
    
    IMPORTANTE: el schema se pasa como JSON example en el prompt,
    no como schema formal. El LLM responde con JSON que se parsea.
    """
    prompt = f"""Eres un analista de documentos técnicos. Extrae la información del siguiente texto y devuélvela como JSON válido.

SCHEMA REQUERIDO (devuelve exactamente esta estructura):
{json.dumps(schema, indent=2, ensure_ascii=False)}

TEXTO DEL DOCUMENTO:
{text}

INSTRUCCIONES:
1. Rellena TODOS los campos del schema. Si no encuentras un dato, usa null para strings/objects y [] para arrays.
2. Para fechas: formato ISO (YYYY-MM-DD). Si solo hay "25 de junio de 2024", convierte a "2024-06-25".
3. Para conclusiones y recomendaciones: texto literal del documento, no resumido.
4. Para arrays (conclusiones, recomendaciones): un elemento por cada uno.
5. Para ubicacion.coordenadas: usa las coordenadas del documento si existen, si no null.
6. Responde SOLO con el JSON, sin texto adicional, sin markdown fences.

JSON:"""

    response = requests.post(
        f"https://api.nan.builders/v1/chat/completions",
        headers={
            "Authorization": f"Bearer {NAN_API}",
            "Content-Type": "application/json"
        },
        json={
            "model": model,
            "messages": [{"role": "user", "content": prompt}],
            "temperature": 0.1,
            "max_tokens": 8192
        },
        timeout=60
    )
    
    result = response.json()
    raw = result["choices"][0]["message"]["content"]
    
    # Limpiar: a veces el LLM envuelve en ```json...```
    raw = raw.strip()
    if raw.startswith("```"):
        raw = raw.split("\n", 1)[1]
    if raw.endswith("```"):
        raw = raw.rsplit("```", 1)[0]
    
    return json.loads(raw)

Fase 4 — Validación

def validate_extraction(data: dict, required_fields: list) -> dict:
    """
    Valida que la extracción tenga los campos requeridos.
    Devuelve: {valid: bool, confidence: float, issues: list}
    """
    issues = []
    score = 0
    total = len(required_fields)
    
    for field in required_fields:
        value = data.get(field)
        if value is None or value == "" or value == []:
            issues.append(f"Campo vacío: {field}")
        elif isinstance(value, str) and value.strip() == "":
            issues.append(f"String vacío: {field}")
        else:
            score += 1
    
    # Validaciones específicas
    if data.get("fecha_suceso"):
        import re
        if not re.match(r'\d{4}-\d{2}-\d{2}', data["fecha_suceso"]):
            issues.append(f"Fecha no en formato ISO: {data['fecha_suceso']}")
            score -= 0.5
    
    if data.get("conclusiones") and len(data["conclusiones"]) == 0:
        issues.append("Array de conclusiones vacío")
    
    confidence = max(0, score / total) if total > 0 else 0
    
    return {
        "valid": len(issues) == 0,
        "confidence": round(confidence * 100, 1),
        "issues": issues
    }

Batch Processing Pattern (producción)

Para procesar muchos PDFs con retry y progress tracking:

import fitz, json, requests, time, os, sys, re
from datetime import datetime

NAN_API = os.getenv("NAN_API")
TEXT_LIMIT = 60000  # Ajustar según análisis de Fase 0

def find_all_pdfs(root_dir):
    """Busca recursivamente TODOS los PDFs — no asumir ubicación."""
    pdfs = []
    for dirpath, _, filenames in os.walk(root_dir):
        for f in filenames:
            if f.lower().endswith('.pdf'):
                pdfs.append(os.path.join(dirpath, f))
    return sorted(pdfs)

def batch_process(pdfs_dir, schema, throttle_ms=2000):
    """Procesa todos los PDFs de un directorio con retry y progress tracking."""
    pdf_files = find_all_pdfs(pdfs_dir)  # ← Buscar recursivamente
    results = []
    errors = []
    total = len(pdf_files)
    start_time = time.time()

    for i, pdf_path in enumerate(pdf_files):
        filename = os.path.basename(pdf_path)
        elapsed = time.time() - start_time
        avg = elapsed / max(i, 1)
        eta = avg * (total - i)
        eta_str = f"~{int(eta//60)}m{int(eta%60):02d}s" if eta > 60 else f"~{int(eta)}s"

        print(f"[{i+1}/{total}] {filename} — {eta_str}", flush=True)

        try:
            # Extract text
            doc = fitz.open(pdf_path)
            full_text = ""
            for page in doc:
                blocks = page.get_text("dict")["blocks"]
                for block in blocks:
                    if block.get("type") != 0:
                        continue
                    for line in block.get("lines", []):
                        for span in line.get("spans", []):
                            full_text += span.get("text") or ""  # ← CRÍTICO: NoneType
            doc.close()

            if len(full_text.strip()) < 100:
                raise Exception(f"Texto insuficiente: {len(full_text)} chars")

            # LLM extraction — truncar a TEXT_LIMIT
            data = llm_extract_structured(full_text[:TEXT_LIMIT], schema)

            # Validate
            validation = validate_extraction(data, list(schema.keys()))
            results.append({
                "file": filename,
                "data": data,
                **validation
            })

        except Exception as e:
            print(f"  ❌ Error: {e}", flush=True)
            errors.append({"file": filename, "error": str(e)})

        time.sleep(throttle_ms / 1000)

    # Save results
    timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M")
    output_path = f"reports_{timestamp}.json"
    with open(output_path, "w") as f:
        json.dump(results, f, ensure_ascii=False, indent=2)

    print(f"\n✅ Procesados: {len(results)}/{total}")
    print(f"❌ Errores: {len(errors)}/{total}")
    print(f"📄 Guardado: {output_path}")
    return results, errors

Reprocesar fallos

def reprocess_failed(errors, pdfs_dir, schema, previous_results):
    """Reintenta PDFs que fallaron (con manejo de NoneType)."""
    new_errors = []
    for error in errors:
        # Reintentar con text=None handling explícito
        # (el fix de span.get("text") or "" ya está en el código principal)
        try:
            # ... retry logic ...
            pass
        except Exception as e:
            new_errors.append({"file": error["file"], "error": str(e)})
    return new_errors

Schema Auto-Learn

Para auto-detectar el schema de un tipo de PDF nuevo:

def auto_learn_schema(pdf_path: str, num_samples: int = 3) -> dict:
    """
    Procesa los primeros N PDFs de un lote para detectar schema.
    
    1. Extrae texto de cada PDF
    2. Envía al LLM con un prompt genérico: "¿Qué campos estructurados ves?"
    3. Consolida los campos comunes
    4. Genera schema JSON
    """
    sample_texts = []
    for i in range(min(num_samples, len(pdf_files))):
        data = extract_text_and_fonts(pdf_files[i])
        text = "\n".join(
            line["text"] 
            for page in data["pages"] 
            for line in page["lines"]
        )[:4000]  # limitar para el prompt
        sample_texts.append(text)
    
    prompt = f"""Analiza estos {num_samples} documentos de tipo similar. 
Identifica los campos estructurados comunes y devuelve un schema JSON.

Documentos:
{chr(10).join(f'--- Documento {i+1} ---{chr(10)}{t[:2000]}' for i, t in enumerate(sample_texts))}

Devuelve un JSON con los campos detectados, cada uno con:
- nombre_campo: tipo_dato
- descripcion: qué representa
- requerido: true/false

JSON:"""
    
    # Llamada LLM...
    return schema

Performance Observada (batch real — v4.0, 270 PDFs)

Comparativa: Regex vs LLM (demostrada con CIAF)

Pitfalls

🔴 LLM fragmenta párrafos en líneas sueltas (POST-EXTRACCIÓN)

PROBLEMA: Después de extraer conclusiones/recomendaciones con LLM, el JSON puede tener cada párrafo partido en líneas individuales de ~60 chars. El 49.8% de los informes CIAF tenía exactamente 20 items (corte del pipeline), cada uno un bullet suelto en vez de un párrafo coherente. En el frontend se ve como una lista de viñetas microscópicas en vez de texto legible.

Detección:

• Promedio de chars/item < 100 (debería ser 200-500 para conclusiones)
• Exactamente 20 items (corte del pipeline LLM)
• Items que no terminan en punto/fin de oración

Solución — script de re-combinación (multi-pass):

def recombine_fragmented_lines(lines):
    """Recombine fragmented lines into coherent paragraphs."""
    paragraphs, current = [], []
    for line in lines:
        line = line.strip()
        if not line: continue
        # Skip metadata/headers
        if re.match(r'^(Informe Final|Comisión de|Subsecretaría|Ministerio)', line): continue
        starts_new = False
        if not current: starts_new = True
        elif re.match(r'^\d+[\.\-\)]\s', line): starts_new = True  # Numbered item
        elif current[-1].endswith('.') and line[0].isupper(): starts_new = True
        elif line[0].islower(): starts_new = False  # Continuation
        elif not current[-1].endswith(('.', '!', '?')): starts_new = False
        else: starts_new = True
        if starts_new and current: paragraphs.append(' '.join(current)); current = [line]
        else: current.append(line)
    if current: paragraphs.append(' '.join(current))
    return [re.sub(r'\s+', ' ', p).strip() for p in paragraphs if len(p) > 10]

Multi-pass limpieza:

1. Pasada 1: Re-combinación de líneas por puntuación y continuidad
2. Pasada 2: Eliminar headers embebidos (5.1. RESUMEN DEL ANÁLISIS, ➢ Conclusiones:, etc.)
3. Pasada 3: Limpiar artefactos (Informe Final XX/XXXX, metadata CIAF, tablas basura)

Resultado CIAF: 194/269 informes actualizados, de 1022 items sueltos a ~600 párrafos coherentes (230 chars/item promedio). Ver references/llm-text-recombination.md para el algoritmo completo con edge cases.

🔴 span["text"] puede ser None en PyMuPDF

CRÍTICO: En algunos PDFs, PyMuPDF retorna None en span["text"] en lugar de string vacío. Causa TypeError: can only concatenate str (not "NoneType") to str en ~25% de PDFs. Solución: SIEMPRE usar span.get("text") or "" o span.get("text", "") or "":

# ❌ ROMPE
line_text += span["text"]

# ✅ SEGURO
line_text += span.get("text") or ""

Este error causó 10 fallos en el batch de 38 CIAF PDFs. Se recuperaron 8/10 reintentando.

🔴 PyMuPDF API changes (v1.24+)

• doc.numPages → len(doc) (propiedad eliminada)
• doc.destroy() → doc.close() (método eliminado)
• Usar siempre doc.close() para liberar memoria

🔴 Python output buffering en background processes

Cuando se ejecuta python3 script.py en background con terminal(background=True), el output se buferiza y no es visible. Solución: PYTHONUNBUFFERED=1 python3 -u script.py o sys.stdout.flush() after each print.

🔴 LLM responses sometimes empty

Algunos PDFs causan respuestas vacías del LLM (posiblemente por contenido tricky o largo). Solución: Verificar content no es None/vacío antes de parsear:

content = result["choices"][0]["message"]["content"] or ""

Si falla, reintentar con texto truncado a 20K chars en vez de 28K.

🔴 Markdown code fences en LLM responses

Qwen 3.6 a veces envuelve JSON en ```json ... ```. SIEMPRE limpiar antes de json.loads():

raw = re.sub(r"^```(?:json)?\s*", "", raw.strip(), flags=re.IGNORECASE)
raw = re.sub(r"\s*```$", "", raw).strip()

🔴 f-strings con JSON braces

Cuando el schema contiene { y }, no se puede interpoliar directamente en f-strings. Usar json.dumps(schema) separado y concatenar.

🔴 Font size clustering: umbral 1.3×

El factor 1.3 funciona bien para documentos técnicos españoles. Para documentos con solo 2 tamaños de fuente (título + body), bajar a 1.15. Para documentos con 4+ niveles de heading, subir a 1.5.

🔴 Chunks grandes saturan el LLM

Si un PDF tiene secciones muy largas (>6000 chars), el LLM puede truncar o ignorar partes. Solución: dividir en chunks de ~4000-6000 chars y procesar por separado, luego consolidar.

🔴 Text limit: analizar ANTES de elegir

NUNCA asumir un límite de texto sin analizar el corpus primero. Los PDFs varían enormemente (13K a 266K chars en CIAF). Si el límite es muy bajo (28K), se trunca información crítica (conclusiones, recomendaciones que están al final del documento). Si es muy alto, se satura el contexto del LLM.

Proceso obligatorio:

1. Extraer texto de 20-50 PDFs representativos
2. Calcular P95 de longitudes
3. Límite = P95 + 5000 margen
4. Si P95 > 80K, considerar enviar solo secciones relevantes (conclusiones, recomendaciones) en vez del texto completo

🔴 Directorio de PDFs: buscar recursivamente

NUNCA asumir que los PDFs están en un directorio conocido. Siempre buscar recursivamente con os.walk() o find. Los PDFs pueden estar distribuidos en subdirectorios por año, categoría, o fuente.

# ❌ ROMPE si hay PDFs en subdirectorios
pdf_files = [f for f in os.listdir(dir) if f.endswith('.pdf')]

# ✅ BUSCA recursivamente
for dirpath, _, filenames in os.walk(root_dir):
    for f in filenames:
        if f.lower().endswith('.pdf'):
            pdf_files.append(os.path.join(dirpath, f))

🔴 API timeout en batch largo

Para batches de 200+ PDFs, es normal encontrar 1-2 timeouts de API (524). No es un error del script — es la API temporalmente caída. Solución: registrar el error y continuar. Los PDFs fallidos se pueden reprocesar después.

🔴 Temperature 0.1, no 0

Temperature 0 causa que el LLM repita exactamente el schema sin rellenar. Temperature 0.1 da consistencia + variación suficiente para llenar campos.

🔴 Rate limiting en batch processing

Para lotes grandes (100+ PDFs), usar throttling de 2s entre requests. Para 1000 PDFs: ~33 minutos con throttling, ~55 minutos sin. Solución: time.sleep(2) entre requests, progress tracking con ETA.

🔴 Multi-schema: PDFs heterogéneos

Para colecciones con múltiples tipos de PDF, crear un schema por tipo. Auto-detectar tipo con las primeras 3-5 muestras usando auto_learn_schema().

🔴 PDFs en directorios inesperados

NUNCA asumir que solo hay un directorio de PDFs. En CIAF, existían 3 ubicaciones:

• ciaf-data/pdfs/ → 38 PDFs (lo que el agente encontró primero)
• CIAF/ → 270 PDFs (la fuente real, con subdirectorios por año)
• CIAF-visor/pdfs/ → copia espejo de CIA/

Solución: Buscar recursivamente en TODO el workspace antes de empezar:

find /root/workspace -name "*.pdf" -not -path "*node_modules*" | wc -l
find /root/workspace -name "*.pdf" -not -path "*node_modules*" | sed 's|/[^/]*$||' | sort | uniq -c | sort -rn

Si el usuario dice "200 PDFs" pero solo encuentras 38, faltan directorios.

🔴 Fusión de datasets con esquemas diferentes

Cuando existen dos fuentes de datos para el mismo conjunto (ej: CIAF-visor con esquema rico + ciaf-data con esquema simple), fusionar selectivamente:

1. Usar el esquema más rico como base (el que tiene más campos: entidades, expediente, GPS, etc.)
2. Mejorar solo los campos débiles del dataset base con los valores del otro
3. NUNCA sobrescribir campos que ya están completos

# Patrón de fusión
for report in base_reports:
    match = find_matching(report, enhanced_data)
    if not match:
        continue
    # Solo mejorar conclusiones/recomendaciones si el otro tiene MÁS
    if len(match.get('conclusiones', [])) > len(report.get('conclusiones', [])):
        report['conclusiones'] = match['conclusiones']
    if len(match.get('recomendaciones', [])) > len(report.get('recomendaciones', [])):
        report['recomendaciones'] = match['recomendaciones']

Resultado CIAF: 62/270 informes mejorados, +111 conclusiones, +59 recomendaciones.

Quality Audit — Auditar datos extraídos

Después de un batch processing, SIEMPRE auditar la calidad de los campos extraídos antes de darlos por buenos.

Auditoría de resúmenes

Un "resumen" extraído de PDFs técnicos suele ser una descripción larga (no un resumen ejecutivo). Detectar y corregir:

import re

def classify_resumen(resumen: str) -> str:
    """Clasifica la calidad de un resumen extraído."""
    if not resumen or len(resumen.strip()) < 20:
        return "empty"
    if len(resumen) > 500:
        return "description"  # ← El 80-95% de los resúmenes caen aquí
    if re.match(r"^El día", resumen, re.IGNORECASE):
        return "description"
    if any(kw in resumen.lower()[:200] for kw in [
        "descripción del suceso", "descripción del accidente",
        "circunstancias del", "el día "
    ]):
        return "description"
    return "summary"  # ← Lo que queremos: 2-3 frases concisas

def audit_quality(reports: list) -> dict:
    """Audita calidad de un lote de informes extraídos."""
    stats = {"total": len(reports), "summary": 0, "description": 0, "empty": 0}
    for r in reports:
        classification = classify_resumen(r.get("resumen", ""))
        stats[classification] += 1
    stats["summary_pct"] = stats["summary"] * 100 // stats["total"]
    return stats

Resultado CIAF (270 informes):

• Antes de re-extract: 16/270 (5%) resúmenes buenos, 254/270 (94%) descripciones
• Después de re-extract: 245/270 (90%) resúmenes buenos

Re-extract de campos deficientes

Cuando un campo tiene mala calidad pero el PDF tiene la información, re-extraer con un prompt específico:

def llm_reextract_resumen(text: str, report: dict, model="qwen3.6") -> str:
    """Re-extrae un resumen ejecutivo de 2-3 frases."""
    truncated = text[:60000]
    
    # Contexto del informe existente
    context = " | ".join(filter(None, [
        f"Tipo: {report['tipo']}" if report.get("tipo") else None,
        f"Lugar: {report['estacion']}" if report.get("estacion") else None,
        f"Fecha: {report['fecha']}" if report.get("fecha") else None,
    ]))

    prompt = f"""Eres un analista de seguridad ferroviaria.
Extrae un RESUMEN EJECUTIVO (MÁXIMO 3 oraciones):
1. Tipo de suceso + ubicación + fecha
2. Consecuencias (víctimas, daños)
3. Causa principal si se menciona

CONTEXTO: {context}
TEXTO: {truncated}

Devuelve SOLO el resumen, sin comillas ni formato."""

    # LLM call...
    return resumen

Cross-repo data matching — Pitfall

Problema: Cuando dos repos tienen datos del mismo conjunto pero con IDs diferentes (ej: 2008-0022/2008 vs IF-0022-2008), el matching por ID no funciona.

Solución: Usar múltiples criterios de matching con scoring:

1. Año (+10 puntos)
2. Número de expediente (extraer dígitos, +20 si coincide)
3. Fecha (+15 si coincide)
4. Estación/ubicación (+5 si una contiene la otra)

def match_reports(base: dict, candidates: list) -> dict:
    """Encuentra el mejor match en otro dataset."""
    best, best_score = None, 0
    for c in candidates:
        score = 0
        if c.get("year") == base.get("year"):
            score += 10
        # Agregar más criterios según los datos disponibles
        if score > best_score:
            best, best_score = c, score
    return best if best_score >= 10 else None

Resultado CIAF: Intento de matching entre ciaf-data y CIAF-visor → 118/269 errores de matching (IDs diferentes). Lección: verificar year+fecha+estación, NO solo ID.

Quality metrics para campos típicos

Batch re-extraction pattern

# Re-extraer solo campos deficientes (no todo el lote)
def reextract_deficient(reports_dir, pdf_dir, field="resumen", 
                        classify_fn=classify_resumen, limit=60000):
    """Re-extrae un campo específico solo para informes deficientes."""
    deficient = []
    for f in os.listdir(reports_dir):
        with open(os.path.join(reports_dir, f)) as fh:
            r = json.load(fh)
        if classify_fn(r.get(field, "")) in ("empty", "description"):
            deficient.append(r)
    
    print(f"Re-extracting {len(deficient)} deficient {field}s...")
    # Procesar con llm_reextract_* y guardar

JSON-to-PDF Verification Audit — Verificar calidad de extracción

CUÁNDO: Después de un batch processing, ANTES de dar los datos por buenos. QUÉ: Comparar cada JSON extraído contra el PDF original para detectar errores de extracción.

Técnica de verificación

Para cada JSON, extraer texto del PDF y verificar 6 campos clave:

import fitz, re

def audit_json_vs_pdf(json_data: dict, pdf_path: str) -> dict:
    """Compara JSON extraído contra texto del PDF original."""
    doc = fitz.open(pdf_path)
    pdf_text = ""
    for page in doc:
        pdf_text += page.get_text()
    doc.close()
    pdf_lower = pdf_text.lower()
    
    results = {}
    
    # 1. TÍTULO — palabras significativas del título deben aparecer en PDF
    titulo_words = [w.lower() for w in json_data.get('titulo', '').split() if len(w) > 4][:8]
    if titulo_words:
        matches = sum(1 for w in titulo_words if w in pdf_lower)
        results['titulo'] = matches / len(titulo_words)
    
    # 2. FECHA — probar múltiples formatos (CRÍTICO: PDFs usan DD/MM/YYYY)
    fecha = json_data.get('fecha', '')
    if fecha:
        parts = fecha.split('-')
        formats = [
            fecha,                              # YYYY-MM-DD
            f"{parts[2]}/{parts[1]}/{parts[0]}", # DD/MM/YYYY
            f"{parts[2]}.{parts[1]}.{parts[0]}", # DD.MM.YYYY
        ]
        date_found = any(fmt in pdf_text for fmt in formats)
        # También probar "DD de mes de YYYY"
        month_names = {'01':'enero','02':'febrero','03':'marzo','04':'abril',
                       '05':'mayo','06':'junio','07':'julio','08':'agosto',
                       '09':'septiembre','10':'octubre','11':'noviembre','12':'diciembre'}
        if not date_found and len(parts) == 3 and parts[1] in month_names:
            text_date = f"{int(parts[2])} de {month_names[parts[1]]} de {parts[0]}"
            date_found = text_date in pdf_lower
        results['fecha'] = 1.0 if date_found else 0.0
    
    # 3. ESTACIÓN — nombre simple debe aparecer
    estacion = json_data.get('estacion', '').split(',')[0].strip()
    if estacion and len(estacion) > 3:
        results['estacion'] = 1.0 if estacion.lower() in pdf_lower else 0.3
    
    # 4. VÍCTIMAS — números clave deben aparecer
    victimas = json_data.get('victimas', 0)
    if victimas > 0:
        results['victimas'] = 1.0 if str(victimas) in pdf_text else 0.4
    else:
        no_casualty = any(p in pdf_lower for p in ['sin víctimas', 'sin fallecidos'])
        results['victimas'] = 1.0 if no_casualty else 0.5
    
    # 5. RESUMEN — palabras del resumen deben estar en PDF (>50% = bueno)
    resumen_words = [w.lower() for w in json_data.get('resumen', '').split() if len(w) > 5][:25]
    if resumen_words:
        coverage = sum(1 for w in resumen_words if w in pdf_lower) / len(resumen_words)
        results['resumen'] = coverage
    
    # 6. CONCLUSIONES — mismo check que resumen
    conc_text = ' '.join(json_data.get('conclusiones', [])).lower()
    conc_words = [w for w in conc_text.split() if len(w) > 5][:25]
    if conc_words:
        coverage = sum(1 for w in conc_words if w in pdf_lower) / len(conc_words)
        results['conclusiones'] = coverage
    
    # Score promedio
    scores = [v for v in results.values() if v > 0]
    avg = sum(scores) / len(scores) if scores else 0
    
    return {
        'fields': results,
        'avg_score': avg,
        'verdict': '✅' if avg >= 0.8 else '⚠️' if avg >= 0.6 else '❌'
    }

Resultados de la auditoría CIAF (270 informes)

Media: 87.3%, Mediana: 88.3% — El pipeline produce datos de alta calidad.

Errores típicos detectados

1. Formato de fecha: JSON YYYY-MM-DD vs PDF DD/MM/YYYY — falsos negativos en verificación directa
2. Campo estación excesivo: Algunos JSONs incluyen PK + tramo + estaciones adyacentes en vez del nombre simple
3. Victimas inconsistentes: Suma de subcategorías > total (ej: fallecidos+graves+leves > victimas)

Dónde están los PDFs de CIAF

/root/workspace/CIAF/          ← FUENTE REAL (277 PDFs, por año)
/root/workspace/CIAF-visor/    ← Copia del visor
/root/workspace/ciaf-data/     ← JSONs extraídos (270)

NUNCA asumir que los PDFs están en ciaf-data/pdfs/ — ese directorio tiene solo 38 PDFs. La fuente completa está en /root/workspace/CIAF/.

Referencias

• references/pipeline-validation-results.md — Resultados del test con 3 informes CIAF 2024
• references/ciaf-schema.json — Schema completo para informes CIAF (11 campos, 100% validado)
• references/batch-processing-notes.md — Notas del batch de 270 PDFs: text limit 60K, NoneType fixes, retry patterns, ETA tracking
• references/quality-audit-patterns.md — Patrones de auditoría de calidad y re-extract
• references/llm-text-recombination.md — Algoritmo de re-combinación de texto fragmentado post-extracción (multi-pass, detección, métricas)
• ocr-quirurgico-pdf-md — Para PDFs escaneados/imágenes (complementario)
• pdf-to-artifacts-david-antizar — Para generación de contenido desde PDFs (complementario)
• markitdown — Para conversión rápida a Markdown sin estructura

PdfToJson v2 — Herramienta general-purpose (refactoring del v1)

El v1 era un script monolítico ligado a CIAF. El v2 (/root/workspace/PdfToJson-v2/) es un módulo Python reutilizable que funciona con cualquier tipo de PDF y múltiples proveedores de LLM.

Arquitectura modular

pdftojson/
├── config.py        → Dataclasses: LLMConfig, ExtractionConfig, OutputConfig
├── extractor.py     → PyMuPDF + chunking inteligente + estimación tokens
├── llm_client.py    → Cliente unificado (Ollama, llama.cpp, vLLM, OpenAI, NaN)
├── orchestrator.py  → Orquestador: process(), process_batch(), merge, save
├── schemas.py       → 5 schemas predefinidos + carga de personalizados
└── __main__.py      → CLI con argparse completo

Qué cambió vs. v1

Multi-backend: patrón cliente unificado

Todos los proveedores usan formato OpenAI-compatible /v1/chat/completions:

# El mismo cliente funciona con todos:
config = LLMConfig(api_url="http://localhost:11434/v1/...", model="qwen2.5:7b")  # Ollama
config = LLMConfig(api_url="http://localhost:8080/v1/...", model="local")          # llama.cpp
config = LLMConfig(api_url="https://api.openai.com/v1/...", model="gpt-4o-mini")  # OpenAI

Pitfall: Ollama usa /v1/chat/completions pero su health check es /api/tags. El cliente detecta el proveedor por URL.

Chunking inteligente

El chunking anterior era un simple text[:60000]. El v2:

1. Calcula el límite automáticamente desde context_window del modelo
2. Corta por párrafos (doble salto de línea) cuando es posible
3. Overlap entre chunks para no perder contexto al cortar
4. Fusión inteligente de resultados: combina arrays, toma strings más largos

# Ejemplo de corte inteligente (de extractor.py):
def _find_best_cut(text, start, end):
    # Prioridad 1: doble salto de línea
    # Prioridad 2: salto de línea
    # Prioridad 3: punto y espacio
    # Fallback: cortar donde esté

Estimación de tokens sin dependencias

# En español, 1 token ≈ 3-4 caracteres. Usamos 3.5 como estimación conservadora.
CHARS_PER_TOKEN = 3.5
def estimate_tokens(text): return int(len(text) / CHARS_PER_TOKEN)
def estimate_chars_for_tokens(n): return int(n * CHARS_PER_TOKEN)

No necesita tiktoken ni transformers. Suficiente para planificación de chunks.

Modelos: contexto vs. calidad

Regla práctica: Para extracción simple (título, fecha, tipo), 7B basta. Para conclusiones largas y recomendaciones estructuradas, 32B+ o API remota.

Uso del CLI

# Análisis previo (sin LLM, solo PyMuPDF)
python -m pdftojojo --analyze *.pdf

# Health check del LLM
python -m pdftojson --health-check --provider ollama

# Procesamiento
python -m pdftojson doc.pdf --schema generico --provider ollama
python -m pdftojson informe.pdf --schema ciaf --provider nan --api-key $NAN_API

# Batch con salida
python -m pdftojson *.pdf --schema generico -o resultados/ --no-consolidated

# Schema personalizado
python -m 

…(truncated)