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({