# PDF LLM Extraction > Extracción inteligente de datos estructurados de PDFs usando análisis de fuentes + LLM. Paradigma: PyMuPDF font analysis → section detection → LLM schema filling → JSON validation. Para PDFs digitales (no escaneados). Validado con 270+ informes CIAF 2024. - Skill: `ntizar/pdf-llm-extraction` (Agent Skill, multi-file: 7 files) - Install (CLI): `npx skillmds@latest add ntizar/pdf-llm-extraction` - Raw SKILL.md: https://api.skillmd.com/api/skills/ntizar/pdf-llm-extraction/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: AI & ML - Author: ntizar (https://skillmd.com/u/ntizar) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/ntizar/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):** ```python # ❌ 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):** ```python # ✅ 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:** ```python 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 ```python 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 ```python 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 ```python 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 ```python 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: ```python 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 ```python 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: ```python 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) | Métrica | Valor | |---------|-------| | Extracción PyMuPDF | ~0.3s por PDF | | Font analysis + sections | ~0.05s por PDF | | LLM structuring (Qwen 3.6) | ~15-20s por PDF | | Validación | ~0.01s por PDF | | **Total por PDF** | **~16-22s** | | **270 PDFs CIAF** | **~71 minutos** (con throttling 1.5s) | | **Tasa de éxito** | **99.6%** (231/232, 1 error API timeout, retry fixeado) | | **Text limit usado** | **60K chars** (P95 del corpus: 143K) | | **max_tokens** | **8192** (suficiente para docs con muchas conclusiones) | ## Comparativa: Regex vs LLM (demostrada con CIAF) | Métrica | Regex (v2.0) | LLM v3.0 (38 PDFs) | LLM v4.0 (270 PDFs) | |---------|--------------|---------------------|----------------------| | Conclusiones extraídas | 16/38 (42%) | 38/38 (100%) | **270/270 (100%)** | | Recomendaciones | 18/38 (47%) | 36/38 (95%) | **268/270 (99%)** | | Trenes identificados | 0/38 (0%) | 38/38 (100%) | **270/270 (100%)** | | Víctimas detectadas | 0 | 200 total | **517 total** | | Texto límite | N/A | 28K chars | **60K chars** | | Texto limpio | Parcial, basura | Literal del informe | **Literal del informe** | | Tiempo/PDF | <1s | ~10-15s | **~16-22s** | | Coste API | $0 | ~$0.01/PDF | **~$0.015/PDF** | | Configuración por tipo | Sí (regex por tipo) | No | **No** | ## 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):** ```python 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 ""`: ```python # ❌ 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: ```python 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()`: ```python 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. ```python # ❌ 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: ```bash 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 ```python # 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: ```python 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**: ```python 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) ```python 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 | Campo | Métrica | Umbral bueno | |-------|---------|-------------| | `resumen` | Longitud | 50-500 chars | | `resumen` | No empieza por "El día" | True | | `conclusiones` | Count | ≥ 2 | | `recomendaciones` | Count | ≥ 1 | | `fecha` | Formato ISO | `YYYY-MM-DD` | | `victimas` | Tipo | int ≥ 0 | ### Batch re-extraction pattern ```python # 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: ```python 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) | Score | Cantidad | % | |-------|----------|---| | EXCELENTE (≥80%) | 244 | 92.4% | | BUENO (60-79%) | 8 | 3.0% | | REGULAR (40-59%) | 11 | 4.2% | | MALO (<40%) | 1 | 0.4% | **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 | v1 (script) | v2 (herramienta) | |---|---| | CIAF-only | Cualquier PDF | | NaN API hardcodeada | 5 proveedores (Ollama, llama.cpp, vLLM, OpenAI, NaN) | | 28-60K chars fixos | Chunking automático según contexto del modelo | | 4 archivos duplicados | 6 módulos limpios, 0 duplicación | | Sin estimación de tokens | Estimación sin dependencias (1 token ≈ 3.5 chars español) | | Sin CLI | CLI completa (--analyze, --health-check, --schema, etc.) | | Paths hardcodeados | Configuración paramétrica | ### Multi-backend: patrón cliente unificado Todos los proveedores usan formato OpenAI-compatible `/v1/chat/completions`: ```python # 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 ```python # 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 ```python # 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 | Modelo | Contexto | Chars caben | Velocidad | Calidad | |---|---|---|---|---| | llama3.1:8b | 8K tokens | ~24K chars | 60-180s (CPU) | Básica | | qwen2.5:7b | 32K tokens | ~96K chars | 10-30s (GPU) | Buena | | qwen2.5:32b | 32K tokens | ~96K chars | 30-60s (GPU) | Muy buena | | gpt-4o-mini | 128K tokens | ~384K chars | 5-15s (API) | Excelente | | qwen3.6 (NaN) | 128K tokens | ~384K chars | 5-15s (API) | Excelente | **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 ```bash # 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)