Generador de JSONs con datos de facturas mediante Llama{Parse,Extract}

00_extract_facturas_llama.py

"""Extracción de datos estructurados de facturas españolas con LlamaExtract.

Este script es un ejemplo *didáctico* del SDK nuevo `llama-cloud` (>= 1.0). Demuestra
cuatro ideas que importan al extraer facturas en producción:

1. EL FLUJO EN TRES ETAPAS
   (1) PARSE    -> client.files.create(...)   sube el PDF y el servicio lo lee (OCR/visión)
   (2) EXTRACT  -> client.extract.run(...)     un LLM rellena nuestro JSON Schema
   (3) VALIDATE -> FacturaEspanola.model_validate(...)  Pydantic valida y normaliza tipos

2. DOS CONTROLES DE CALIDAD, EN DOS FASES DISTINTAS
   - parse_tier : calidad de la fase (1), cómo se LEE el PDF. 'agentic' lee mejor tablas
                  y escaneos difíciles.
   - tier       : calidad de la fase (2), el LLM que RELLENA el esquema.
                  Valores: cost_effective | agentic.

3. ESCALADO AUTOMÁTICO
   Empezamos barato (cost_effective). Si la extracción sale VACÍA (señal de que el PDF
   no se pudo leer bien), reintentamos automáticamente subiendo AMBOS controles a
   'agentic'. No escalamos ante cualquier error: solo cuando el resultado es basura,
   porque hay fallos (p.ej. un campo nulo) que no se arreglan con más cómputo.

4. INYECCIÓN DE CONTEXTO POR FICHERO (ver FILE_RULES)
   El LLM acepta un `system_prompt` (instrucciones en lenguaje natural). Además de un
   prompt base común, podemos AÑADIR contexto específico según el nombre del fichero
   —p.ej. "esta factura anonimiza datos con asteriscos, es esperado"— y ajustar
   `target_pages` para saltarnos páginas irrelevantes (condiciones, publicidad...).

Uso:
    uv run extract_facturas.py                          # facturas de ejemplo por defecto
    uv run extract_facturas.py --all                    # todos los .pdf de invoices/
    uv run extract_facturas.py --tier agentic f.pdf     # forzar tier alto
    uv run extract_facturas.py --target-pages 1 f.pdf   # limitar páginas manualmente

La API key se lee de LLAMA_PARSE_KEY (o LLAMA_CLOUD_API_KEY) en el .env. El SDK solo
entiende LLAMA_CLOUD_API_KEY, así que la mapeamos al arrancar.
"""

from __future__ import annotations

import argparse
import json
import os
import sys
from pathlib import Path

from dotenv import load_dotenv

from schema import FacturaEspanola

# Facturas "rellenas" de ejemplo (datos reales) que mejor demuestran la extracción.
DEFAULT_FILES = [
    "invoices/xxxxxxxxxxxxxxxxx.pdf",
    "invoices/yyyyyyyyyyyyyyyyy.pdf",
]

# ── Costes (para el bloque metadata) ─────────────────────────────────────────────────
# La API NO devuelve los créditos consumidos: el consumo real vive en el dashboard
# (Usage), localizable por el extract_job_id (ext-...). Aquí estimamos el coste con el
# modelo de tarifas POR PÁGINA = parse + extract. Como extract.run() hace ambas fases,
# sumamos las dos. Ajusta estas tarifas si cambian en tu cuenta/región.
PRICE_PER_1000_CREDITS_USD = 1.25     # tarifa oficial LlamaCloud (NA/UE)
USD_TO_EUR = 0.92                     # conversión aproximada
PARSE_CREDITS_PER_PAGE = {"normal": 3, "agentic": 10}      # fase (1) según parse_tier
EXTRACT_CREDITS_PER_PAGE = {"cost_effective": 5, "agentic": 15}  # fase (2) según tier


def pdf_page_count(path: str) -> int:
    """Número de páginas del PDF (1 como fallback si no se puede leer)."""
    try:
        from pypdf import PdfReader
        return len(PdfReader(path).pages) or 1
    except Exception:  # noqa: BLE001 — el coste estimado no debe romper la extracción
        return 1


def pages_in_spec(target_pages: str) -> int:
    """Cuenta páginas de una spec tipo '1', '1,2' o '1-3' (para target_pages)."""
    pages: set[int] = set()
    for tok in target_pages.split(","):
        tok = tok.strip()
        if "-" in tok:
            a, b = tok.split("-", 1)
            pages.update(range(int(a), int(b) + 1))
        elif tok:
            pages.add(int(tok))
    return len(pages) or 1


def effective_pages(pdf: str, target_pages: str | None) -> int:
    """Páginas que se facturarán: las de target_pages si se limita, si no, todo el PDF."""
    return pages_in_spec(target_pages) if target_pages else pdf_page_count(pdf)


def attempt_credits(tier: str, parse_tier: str | None, pages: int) -> int:
    """Créditos estimados de UN intento = (parse + extract) por página × nº de páginas."""
    parse_kind = "agentic" if parse_tier == "agentic" else "normal"
    per_page = PARSE_CREDITS_PER_PAGE[parse_kind] + EXTRACT_CREDITS_PER_PAGE.get(tier, 0)
    return per_page * pages


def credits_to_money(credits: int) -> dict:
    """Convierte créditos a coste estimado en USD y EUR."""
    usd = credits * PRICE_PER_1000_CREDITS_USD / 1000
    return {"usd": round(usd, 4), "eur": round(usd * USD_TO_EUR, 4)}

# Prompt base que acompaña SIEMPRE a la extracción. No es código: es lenguaje natural
# que el modelo lee junto al esquema para resolver ambigüedades típicas de las facturas
# españolas. Las reglas por fichero (FILE_RULES) se AÑADEN a este prompt, no lo sustituyen.
SYSTEM_PROMPT = (
    "Estás extrayendo datos de una factura conforme a la normativa española. "
    "Reglas: "
    "1) Devuelve importes como números planos, sin símbolo de moneda y con punto decimal. "
    "2) En facturas intracomunitarias o con inversión del sujeto pasivo (reverse charge) "
    "el IVA puede ser 0: refléjalo tal cual, no lo inventes. "
    "3) Ignora páginas de condiciones legales, publicidad o detalle de consumos que no "
    "formen parte de los importes facturados. "
    "4) Copia el NIF/CIF/DNI de emisor y receptor exactamente como aparecen. "
    "5) Si un dato no aparece, déjalo vacío en lugar de inventarlo."
)

# ── Reglas de inyección de contexto por fichero ──────────────────────────────────────
# Cada regla es un diccionario:
#   match        : lista de subcadenas (en minúsculas). La regla se aplica si CUALQUIERA
#                  aparece en el nombre del fichero (case-insensitive).
#   note         : texto que se AÑADE al system_prompt cuando la regla casa.
#   target_pages : (opcional) override de páginas a procesar, p.ej. "1,2".
#   tier         : (opcional) override del tier de arranque para ese fichero.
# Añadir un emisor nuevo es tan simple como añadir una entrada aquí.
FILE_RULES: list[dict] = [
    {
        "match": ["naturgy"],
        "note": (
            "Esta es una factura de Naturgy (gas/luz): suele traer páginas de detalle "
            "de consumo, gráficas y condiciones que NO forman parte de los importes. "
            "Quédate solo con el resumen de la factura de las primeras páginas."
        ),
        "target_pages": "1,2",  # nos saltamos el detalle de consumo posterior
    },
    {
        "match": ["o2", "telefon", "om7vafj"],
        "note": (
            "En esta factura (O2 / Telefónica) muchos datos aparecen ANONIMIZADOS con "
            "asteriscos (p.ej. 'B7***7191' o 'PETALO VEN********'). Esto es esperado y "
            "NO es un error: copia los valores tal cual aparecen, con sus asteriscos, "
            "sin intentar reconstruir el dato completo."
        ),
    },
]


def overrides_for(path: str) -> dict:
    """Calcula el contexto/overrides aplicables a un fichero según FILE_RULES.

    Recorre FILE_RULES y acumula lo que aporten todas las reglas cuyo `match` aparezca
    en el nombre del fichero. Si dos reglas definen `target_pages`/`tier`, gana la última.

    Args:
        path: ruta (o nombre) del PDF. Solo se mira el nombre del fichero, en minúsculas.

    Returns:
        dict con cuatro claves:
            - note (str)        : notas concatenadas a añadir al system_prompt ("" si ninguna).
            - target_pages (str|None): override de páginas (p.ej. "1,2") o None.
            - tier (str|None)   : override del tier de arranque o None.
            - applied (list[str]): etiquetas de las reglas que dispararon (para log).
    """
    name = Path(path).name.lower()
    notes: list[str] = []
    target_pages: str | None = None
    tier: str | None = None
    applied: list[str] = []
    for rule in FILE_RULES:
        if any(token in name for token in rule["match"]):
            notes.append(rule["note"])
            target_pages = rule.get("target_pages", target_pages)
            tier = rule.get("tier", tier)
            applied.append(rule["match"][0])  # etiqueta legible de la regla
    return {
        "note": "\n".join(notes),
        "target_pages": target_pages,
        "tier": tier,
        "applied": applied,
    }


def configure_api_key() -> str:
    """Carga el .env y normaliza la API key al nombre que el SDK espera.

    Acepta la key en LLAMA_PARSE_KEY o en LLAMA_CLOUD_API_KEY (esta última tiene
    prioridad). Sea cual sea, la deja exportada en LLAMA_CLOUD_API_KEY, que es la única
    variable que lee `llama_cloud`. Aborta el programa si no encuentra ninguna.

    Args:
        None. Lee del entorno / fichero .env del directorio actual.

    Returns:
        La API key (str). Side effect: define os.environ["LLAMA_CLOUD_API_KEY"].

    Raises:
        SystemExit: si no hay ninguna key definida.

    --- .env
    # API key de LlamaCloud (https://cloud.llamaindex.ai → API Keys). Empieza por "llx-".
    LLAMA_CLOUD_API_KEY=llx-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx

    """
    load_dotenv()
    key = os.environ.get("LLAMA_CLOUD_API_KEY")
    if not key:
        sys.exit(
            "ERROR: no se encontró la API key. Define LLAMA_PARSE_KEY "
            "(o LLAMA_CLOUD_API_KEY) en tu .env."
        )
    os.environ["LLAMA_CLOUD_API_KEY"] = key
    return key


def looks_empty(f: FacturaEspanola) -> bool:
    """Decide si una extracción válida está, en la práctica, "vacía" (basura).

    Una factura puede pasar la validación de Pydantic y aun así no contener datos útiles
    (todo strings vacíos y total 0). Eso suele indicar que el parser no pudo LEER el PDF
    —típico de escaneos malos en cost_effective—, y es la señal que dispara el escalado a
    'agentic'. No escalamos ante cualquier error, solo ante este "vacío".

    Args:
        f: la FacturaEspanola ya validada que queremos evaluar.

    Returns:
        True si no hay ni número de factura ni total (candidata a reintento); False si
        parece tener contenido real.
    """
    return not (f.numero_factura or "").strip() and not f.total_factura


def print_resumen(factura: FacturaEspanola) -> None:
    """Imprime por consola un resumen legible de los campos clave de la factura.

    Solo muestra la retención de IRPF si existe (facturas de autónomo). Es puramente
    informativo: no devuelve nada ni altera la factura.

    Args:
        factura: la FacturaEspanola validada a resumir.

    Returns:
        None. Escribe en stdout.
    """
    print(f"  Nº factura     : {factura.numero_factura}")
    print(f"  Fecha          : {factura.fecha_expedicion}")
    print(f"  Emisor         : {factura.emisor.nombre}  ({factura.emisor.nif})")
    print(f"  Receptor       : {factura.receptor.nombre}  ({factura.receptor.nif})")
    print(f"  Base imponible : {factura.base_imponible_total}")
    print(f"  Total IVA      : {factura.total_iva}")
    if factura.retencion_irpf_importe:
        print(f"  Retención IRPF : {factura.retencion_irpf_importe}")
    print(f"  TOTAL          : {factura.total_factura} {factura.moneda or ''}")


def main() -> None:
    """Punto de entrada: parsea argumentos y procesa el lote de facturas.

    Orquesta todo el flujo por cada PDF: aplica reglas por fichero (overrides_for),
    construye la escalera de intentos (cost_effective → agentic si procede), ejecuta
    PARSE→EXTRACT→VALIDATE, vuelca el JSON crudo y el validado, e imprime progreso y un
    resumen final de OK/fallos. Un fallo en una factura no aborta el resto del lote.

    Args:
        None. Lee la configuración de la línea de comandos (ver --help) y del .env.

    Returns:
        None. Efectos: escribe ficheros JSON en --out e imprime el progreso en stdout.
    """
    parser = argparse.ArgumentParser(description="Extrae datos de facturas españolas con LlamaExtract")
    parser.add_argument("files", nargs="*", default=DEFAULT_FILES, help="PDFs a procesar")
    parser.add_argument(
        "--all", action="store_true", help="Procesa todos los .pdf de invoices/"
    )
    parser.add_argument(
        "--tier",
        default="cost_effective",
        choices=["cost_effective", "agentic"],
        help="Tier de la fase EXTRACT (el LLM). agentic = más calidad y coste.",
    )
    parser.add_argument(
        "--no-escalate",
        action="store_true",
        help="Desactiva el reintento automático a 'agentic' cuando la extracción sale vacía.",
    )
    parser.add_argument(
        "--target-pages",
        default=None,
        help="Limita a estas páginas (p.ej. '1' o '1,2'). Una regla por fichero puede sobreescribirlo.",
    )
    parser.add_argument(
        "--system-prompt",
        default=SYSTEM_PROMPT,
        help="Prompt base para guiar al modelo (cadena vacía = ninguno). Las reglas por fichero se le añaden.",
    )
    parser.add_argument(
        "--out", default="output", help="Carpeta donde volcar los JSON resultantes"
    )
    args = parser.parse_args()

    api_key = configure_api_key()

    # Importamos aquí para que el --help funcione aunque falten credenciales.
    from llama_cloud import LlamaCloud

    if args.all:
        candidates = [str(p) for p in sorted(Path("invoices").glob("*.pdf"))]
    else:
        candidates = args.files or DEFAULT_FILES
    files = [f for f in candidates if Path(f).is_file()]
    if not files:
        sys.exit("No se encontró ningún PDF que procesar.")

    out_dir = Path(args.out)
    out_dir.mkdir(exist_ok=True)

    client = LlamaCloud(api_key=api_key)

    # El esquema Pydantic se convierte a JSON Schema: esto es lo que el LLM recibe como
    # "molde" a rellenar (los Field(description=...) viajan dentro como pistas).
    data_schema = FacturaEspanola.model_json_schema()

    def build_config(tier: str, parse_tier: str | None, system_prompt: str,
                     target_pages: str | None) -> dict:
        """Construye el dict `configuration` para client.extract.run().

        Captura `data_schema` del scope de main(). Los campos opcionales solo se añaden
        si tienen valor, para no enviar claves vacías a la API.

        Args:
            tier: tier de la fase EXTRACT ("cost_effective" | "agentic").
            parse_tier: tier de la fase PARSE, o None para usar el por defecto del servicio.
            system_prompt: contexto en lenguaje natural (cadena vacía = no enviar ninguno).
            target_pages: subconjunto de páginas (p.ej. "1,2"), o None para todas.

        Returns:
            dict listo para pasar como `configuration=` a client.extract.run().
        """
        cfg: dict = {
            "data_schema": data_schema,
            "extraction_target": "per_doc",  # un objeto por documento (no por página)
            "tier": tier,                     # calidad de la fase EXTRACT
        }
        if system_prompt:
            cfg["system_prompt"] = system_prompt   # contexto en lenguaje natural
        if target_pages:
            cfg["target_pages"] = target_pages     # subconjunto de páginas
        if parse_tier:
            cfg["parse_tier"] = parse_tier         # calidad de la fase PARSE (lectura)
        return cfg

    total = len(files)
    print(f"\nTier base: {args.tier}  |  Escalado: {'no' if args.no_escalate else 'sí'}"
          f"  |  Facturas: {total}\n" + "=" * 60)

    ok, failed = 0, []
    total_credits = 0      # acumulado de TODOS los intentos (también los que fallaron)
    escalated_docs = 0     # nº de facturas que necesitaron escalar
    for i, pdf in enumerate(files, start=1):
        doc_i = f"[{i}/{total}]"
        stem = Path(pdf).stem
        print(f"\n▶ {doc_i} {pdf}", flush=True)

        # Reglas por fichero: contexto extra + posibles overrides de páginas/tier.
        ov = overrides_for(pdf)
        system_prompt = args.system_prompt
        if ov["note"]:
            system_prompt = (system_prompt + "\n" + ov["note"]).strip()
            print(f"  ↳ contexto inyectado por regla: {', '.join(ov['applied'])}", flush=True)
        target_pages = ov["target_pages"] or args.target_pages
        if ov["target_pages"]:
            print(f"  ↳ páginas limitadas a: {ov['target_pages']}", flush=True)
        start_tier = ov["tier"] or args.tier
        pages = effective_pages(pdf, target_pages)  # nº de páginas que se facturan

        # Escalera de intentos: tier de arranque; si sale vacío y procede, 'agentic'
        # subiendo TAMBIÉN parse_tier (mejor lectura del PDF).
        attempts = [(start_tier, None)]
        if not args.no_escalate and start_tier != "agentic":
            attempts.append(("agentic", "agentic"))
        # Fases lógicas SIEMPRE 3: Parse · Extract · Validate. El escalado es un REINTENTO
        # dentro de la fase Extract (se muestra como "↳ reintento"), no una 4ª fase.
        N = 3

        # (1) PARSE: se sube/parsea UNA vez; se reutiliza en todos los intentos.
        try:
            print(f"  {doc_i} (paso 1/{N}) Parse (subir+OCR)…", flush=True)
            uploaded = client.files.create(file=Path(pdf), purpose="extract")
        except Exception as err:  # noqa: BLE001
            print(f"  ⚠ {doc_i} Falló al subir: {type(err).__name__}: {err}")
            failed.append(pdf)
            continue

        # (2) EXTRACT + (3) VALIDATE, con escalado. Vamos registrando CADA intento
        # (su job id y coste estimado) para poder auditar después incluso el intento
        # barato que falló antes de escalar.
        factura, last_err, winning_job = None, None, None
        intentos_log: list[dict] = []
        for a, (tier, parse_tier) in enumerate(attempts):
            etiqueta = tier + (" + parse agentic" if parse_tier else "")
            if a == 0:
                print(f"  {doc_i} (paso 2/{N}) Extract [{etiqueta}]…", flush=True)
            else:
                # Reintento por escalado: NO es una fase nueva, es la misma fase Extract.
                print(f"      ↳ reintento {a + 1}: Extract [{etiqueta}]…", flush=True)
            job = None  # si extract.run() falla, no hay job id para este intento
            try:
                job = client.extract.run(
                    file_input=uploaded.id,
                    configuration=build_config(tier, parse_tier, system_prompt, target_pages),
                )
                raw = job.extract_result
                # Volcamos SIEMPRE el JSON crudo: diagnóstico sin re-llamar a la API.
                (out_dir / f"{stem}.raw.json").write_text(
                    json.dumps(raw, ensure_ascii=False, indent=2), encoding="utf-8"
                )
                candidato = FacturaEspanola.model_validate(raw)
            except Exception as err:  # noqa: BLE001
                last_err = err
                intentos_log.append({
                    "extract_job_id": getattr(job, "id", None),
                    "tier": tier, "parse_tier": parse_tier,
                    "resultado": "error",
                    "creditos_estimados": attempt_credits(tier, parse_tier, pages),
                })
                continue  # probamos el siguiente intento (si lo hay)

            vacio = looks_empty(candidato)
            intentos_log.append({
                "extract_job_id": job.id,                 # el ext-... de ESTE intento
                "tier": tier, "parse_tier": parse_tier,
                "resultado": "vacío" if vacio else "ok",
                "creditos_estimados": attempt_credits(tier, parse_tier, pages),
            })

            # Si salió vacío y aún quedan intentos, escalamos.
            if vacio and a < len(attempts) - 1:
                print("      ↑ extracción vacía, escalando tier…", flush=True)
                continue
            factura, winning_job = candidato, job
            break

        # Acumulamos coste de este documento (intentos que SÍ se ejecutaron), tanto si
        # acabó OK como si falló: el gasto se produce igual.
        total_credits += sum(it["creditos_estimados"] for it in intentos_log)
        if len(intentos_log) > 1:
            escalated_docs += 1

        if factura is None:
            tipo = type(last_err).__name__ if last_err else "ExtracciónVacía"
            print(f"  ⚠ {doc_i} Falló: {tipo}")
            for line in str(last_err or "sin datos extraíbles").splitlines():
                print(f"      {line}")
            print(f"      (JSON crudo en {out_dir / f'{stem}.raw.json'})")
            failed.append(pdf)
            continue

        # ── metadata: trazabilidad + coste estimado (la API no devuelve créditos) ──
        creditos_totales = sum(it["creditos_estimados"] for it in intentos_log)
        ganador = intentos_log[-1]  # el último intento es el que tuvo éxito
        metadata = {
            "source_pdf": pdf,
            "file_id": uploaded.id,                       # UUID del fichero subido
            "project_id": getattr(winning_job, "project_id", None),
            "extract_job_id": winning_job.id,             # ext-... del intento bueno
            "status": getattr(winning_job, "status", None),
            "parsing_nota": "extract.run() parsea internamente; no expone un parse job id (pjb-...).",
            "tier": ganador["tier"],
            "parse_tier": ganador["parse_tier"],
            "target_pages": target_pages,
            "paginas_facturadas": pages,
            "reglas_aplicadas": ov["applied"],
            "escalado": len(intentos_log) > 1,            # ¿hubo reintento caro?
            "num_intentos": len(intentos_log),
            "intentos": intentos_log,                     # cada job id + tier + coste
            "coste_estimado": {
                "creditos": creditos_totales,
                **credits_to_money(creditos_totales),
                "nota": "Estimación por tier; el consumo real está en el dashboard (Usage).",
            },
        }

        out_path = out_dir / f"{stem}.json"
        out_path.write_text(
            json.dumps(
                {"metadata": metadata, "factura": factura.model_dump()},
                ensure_ascii=False, indent=2, default=str,
            ),
            encoding="utf-8",
        )
        print(f"  {doc_i} (paso 3/{N}) Validate OK", flush=True)
        print_resumen(factura)
        coste = metadata["coste_estimado"]
        marca = "  ⚠ ESCALADO" if metadata["escalado"] else ""
        print(f"  ⓘ job {winning_job.id} · tier {ganador['tier']} · {pages} pág"
              f" · ~{coste['creditos']} cr (~€{coste['eur']}){marca}")
        print(f"  → JSON         : {out_path}")
        ok += 1

    print("\n" + "=" * 60)
    print(f"Listo: {ok} OK, {len(failed)} con fallo. JSON en ./{out_dir}/")
    dinero = credits_to_money(total_credits)
    procesadas = ok + len(failed)
    media = total_credits / procesadas if procesadas else 0
    print(f"Coste estimado total: ~{total_credits} cr"
          f"  ·  ~${dinero['usd']}  ·  ~€{dinero['eur']}")
    print(f"  ({procesadas} facturas · media ~{media:.1f} cr/factura · {escalated_docs} escaladas)")
    print("  Estimación con tarifas por página; el consumo real está en el dashboard (Usage).")
    if failed:
        print("Fallaron:")
        for f in failed:
            print(f"  - {f}")


if __name__ == "__main__":
    main()

README.md

Extracción de facturas españolas con LlamaExtract

Ejemplo mínimo y reproducible de cómo usar LlamaExtract (LlamaCloud) para sacar datos estructurados de facturas en PDF, con un esquema modelado sobre el formato legal español (NIF/CIF, base imponible, desglose de IVA por tipo, retención de IRPF…).

Más allá del "hola mundo", el ejemplo muestra cómo hacerlo robusto: escalado automático de calidad cuando un PDF no se lee bien, e inyección de contexto por fichero para casos peculiares (facturas con páginas extra, datos anonimizados, etc.).

Cómo funciona

No es un "resumen": es generación restringida. El esquema no solo valida, sino que forma parte del prompt — LlamaCloud lo convierte a JSON Schema y obliga al LLM a rellenar esa estructura exacta. El flujo (extract_facturas.py) tiene tres etapas explícitas:

PDF ──(1) PARSE──► texto+layout ──(2) EXTRACT (LLM)──► JSON ──(3) VALIDATE──► Pydantic
    files.create()                 extract.run()             model_validate()

1. Parse — client.files.create(file=pdf, purpose="extract") sube el PDF y el servicio lo lee (OCR/visión).
2. Extract — client.extract.run(...) pasa tu JSON Schema al LLM, que devuelve un JSON que lo cumple. Aquí cada Field(description=...) de schema.py actúa como instrucción de qué buscar (NIF/CIF, base imponible, IRPF…).
3. Validate — FacturaEspanola.model_validate(...) valida tipos y normaliza el JSON crudo a un objeto Pydantic.

Por eso devuelve null en vez de inventar: si no encuentra el IRPF, el esquema lo permite y el modelo lo respeta.

Dos controles de calidad (en dos fases distintas)

La "calidad" no es un único ajuste. Hay dos controles, uno por fase:

Config	Fase	Qué controla	Valores
parse_tier	(1) Parse	Cómo se lee el PDF (OCR/layout). agentic lee mejor tablas y escaneos difíciles.	cost_effective, agentic
tier	(2) Extract	El LLM que rellena el esquema.	cost_effective, agentic

cost_effective es rápido y barato; agentic es más lento y caro, pero entiende documentos complejos. Empezamos barato y subimos solo cuando hace falta (ver abajo).

El proceso de decisión

Por cada factura, el script decide cuánto contexto dar y cuánta potencia gastar:

                    ┌─ ¿el nombre casa una regla? (FILE_RULES)
                    │     sí → añade contexto al system_prompt
 factura.pdf ───────┤            y/o limita target_pages
                    │
                    └─► PARSE (1 vez) ─► EXTRACT [cost_effective] ─► VALIDATE
                                                      │
                                          ¿salió vacía (sin nº ni total)?
                                             │ sí                    │ no
                                             ▼                       ▼
                                  EXTRACT [agentic+parse agentic]   ✅ OK
                                             │
                                          VALIDATE ─► ✅ OK / ⚠ fallo

Claves del diseño:

• Escalado solo ante extracción vacía. No subimos el tier ante cualquier error. Hay fallos (p.ej. el modelo devolvió un campo null que el esquema rechazaba) que no se arreglan con más cómputo, sino corrigiendo el contrato de datos. Solo escalamos cuando el resultado es basura (ni número de factura ni total) — señal de que el PDF no se pudo leer, que es justo lo que agentic mejora.
• El PDF se sube una sola vez. Los reintentos reutilizan file_input=uploaded.id, así escalar solo paga la fase de extract, no re-subir el documento.
• Se vuelca SIEMPRE el JSON crudo (output/<factura>.raw.json) antes de validar. Si algo falla, se diagnostica sin volver a llamar a la API (y gastar créditos).

Inyección de contexto por fichero (FILE_RULES)

El LLM acepta un system_prompt: instrucciones en lenguaje natural que se leen junto al esquema. Además del prompt base (formato de importes, reverse charge, ignorar páginas de condiciones…), se puede añadir contexto específico según el nombre del fichero, y ajustar las páginas a procesar. Las reglas viven en FILE_RULES:

FILE_RULES = [
    {   # Naturgy trae páginas de detalle de consumo que estorban
        "match": ["naturgy"],
        "note": "...quédate solo con el resumen de las primeras páginas...",
        "target_pages": "1,2",          # nos saltamos el detalle posterior
    },
    {   # O2/Telefónica anonimiza datos con asteriscos
        "match": ["o2", "telefon", "om7vafj"],
        "note": "...los asteriscos (B7***7191) son esperados, NO un error: cópialos tal cual...",
    },
]

• match: la regla se aplica si cualquiera de esas subcadenas está en el nombre del fichero (case-insensitive).
• note: se añade al system_prompt base (no lo sustituye).
• target_pages / tier: overrides opcionales de páginas y de tier de arranque.

Añadir un emisor nuevo es una entrada más en la lista — cero cambios en la lógica.

Requisitos

• Una API key de LlamaCloud (https://cloud.llamaindex.ai → API Keys, empieza por llx-).
• uv para gestionar el entorno.

cp .env.example .env        # y pon tu key en LLAMA_PARSE_KEY
uv sync                     # instala dependencias

El SDK lee la variable LLAMA_CLOUD_API_KEY. Como en este proyecto la key está en LLAMA_PARSE_KEY, el script la mapea automáticamente al arrancar.

Uso

# Facturas de ejemplo por defecto
uv run extract_facturas.py

# Todas las de invoices/
uv run extract_facturas.py --all

# Una factura concreta
uv run extract_facturas.py invoices/factura-simple-iva21.pdf

# Forzar tier alto desde el principio (sin esperar al escalado)
uv run extract_facturas.py --tier agentic invoices/factura_o2.pdf

# Limitar páginas manualmente (una regla por fichero puede sobreescribirlo)
uv run extract_facturas.py --target-pages 1 "invoices/Petalo Uber Mayo.pdf"

# Desactivar el reintento automático a agentic
uv run extract_facturas.py --all --no-escalate

Flags principales: --all, --tier {cost_effective,agentic}, --no-escalate, --target-pages, --system-prompt, --out. Cada factura genera en output/ un JSON con la forma { "metadata": {...}, "factura": {...} } (ver más abajo), más un <factura>.raw.json con la respuesta cruda del modelo antes de validar.

Salida en consola

Reporta el progreso en tiempo real: documento [n/total], paso (k/total_pasos), qué reglas se inyectaron y si hubo escalado. Ejemplo real (O2, que escala):

Las fases lógicas son siempre 3 (Parse · Extract · Validate); el escalado se muestra como un reintento dentro de la fase Extract, no como una fase nueva:

▶ [2/2] invoices/factura_o2.pdf
  ↳ contexto inyectado por regla: o2
  [2/2] (paso 1/3) Parse (subir+OCR)…
  [2/2] (paso 2/3) Extract [cost_effective]…
      ↑ extracción vacía, escalando tier…
      ↳ reintento 2: Extract [agentic + parse agentic]…
  [2/2] (paso 3/3) Validate OK
  Nº factura     : OM7VAFJ001****
  Emisor         : Telefónica de España, S.A.U.  (A-82018474)
  TOTAL          : 43.0 EUR
  ⓘ job ext-wsi9nykka75…apw · tier agentic · 2 pág · ~66 cr (~€0.0759)  ⚠ ESCALADO
  → JSON         : output/factura_o2.json

Y al final del lote, un agregado de coste para no leer microcéntimos uno a uno:

============================================================
Listo: 9 OK, 0 con fallo. JSON en ./output/
Coste estimado total: ~178 cr  ·  ~$0.2225  ·  ~€0.2047
  (9 facturas · media ~19.8 cr/factura · 1 escaladas)
  Estimación con tarifas por página; el consumo real está en el dashboard (Usage).

Estructura del JSON de salida

Cada output/<factura>.json envuelve los datos en dos bloques:

{
  "metadata": {
    "extract_job_id": "ext-…",        // localizable en el dashboard (Usage)
    "file_id": "…", "project_id": "…",
    "tier": "agentic", "parse_tier": "agentic",
    "target_pages": null, "paginas_facturadas": 2,
    "reglas_aplicadas": ["o2"],
    "escalado": true, "num_intentos": 2,
    "intentos": [                       // CADA intento, incluido el barato que falló
      {"extract_job_id": "ext-…", "tier": "cost_effective", "resultado": "vacío", "creditos_estimados": 16},
      {"extract_job_id": "ext-…", "tier": "agentic", "parse_tier": "agentic", "resultado": "ok", "creditos_estimados": 50}
    ],
    "coste_estimado": {"creditos": 66, "usd": 0.0825, "eur": 0.0759, "nota": "…dashboard…"},
    "parsing_nota": "extract.run() parsea internamente; no expone un parse job id (pjb-…)."
  },
  "factura": { "numero_factura": "…", "emisor": {...}, "desglose_iva": [...], "total_factura": 43.0, ... }
}

El metadata da trazabilidad (qué job, qué tier, qué reglas) y coste; el bloque factura es el objeto Pydantic validado. Nota: el SDK no devuelve un parse job id (pjb-…) porque extract.run() parsea internamente dentro del mismo job ext-….

<factura>.json vs <factura>.raw.json

Por cada PDF se escriben dos ficheros en output/. Son las dos caras de la etapa VALIDATE (la entrada y la salida), no duplicados:

Fichero	Qué es	Estructura	Cuándo se escribe
<factura>.raw.json	La respuesta cruda del modelo (job.extract_result), antes de validar	Plana (campos de la factura, tal cual). Puede traer null donde el esquema espera lista, tipos sin normalizar, etc.	Siempre, aunque la validación falle después
<factura>.json	El resultado final validado	{ "metadata": {...}, "factura": {...} }. El bloque factura ya pasó por Pydantic (null → [], str → float, defaults) y se le añade metadata (job ids, tier, coste)	Solo si la validación tiene éxito

Ejemplo real: en Petalo Uber Mayo, el modelo devolvió lineas: null (sin tabla de líneas). El .raw.json lo conserva como null; el .json lo muestra ya coercionado a [] por el validador de schema.py.

Para qué sirve cada uno:

• .json es el dato que consume tu aplicación (validado, tipado, con metadata).
• .raw.json es para depurar sin coste: si una factura falla la validación, el .json no se escribe, pero el .raw.json sí — así ves qué devolvió el modelo y por qué Pydantic lo rechazó, sin re-llamar a la API (sin gastar créditos). En producción normalmente se descartan; aquí se conservan por ser un ejemplo didáctico.

Costes

La API no devuelve los créditos consumidos (no hay endpoint de usage en el SDK); el consumo real vive en el dashboard, localizable por extract_job_id. El script estima el coste con el modelo de tarifas por página = parse + extract:

Fase	cost_effective	agentic
Parse	3 cr/pág	10 cr/pág
Extract	5 cr/pág	15 cr/pág
Total por página	8 cr	25 cr

A $1,25 / 1.000 créditos (~€0,00115/crédito). Implicaciones:

• Una factura de 1 página en barato cuesta ~8 cr (~€0,009). Multiplica por páginas: una de 4 páginas son ~32 cr.
• Escalar es caro y asimétrico: pagas el intento barato fallido y el agentic. Una de 4 páginas que escala salta de 32 a 132 cr (~4×).
• target_pages ahorra de verdad: Naturgy tiene 4 páginas pero la regla la limita a 2 → la mitad de coste, sin perder datos de la factura.
• El agregado final suma todos los intentos (también los de facturas que fallaron, porque se cobran igual) y reporta la media y cuántas escalaron.

Ajusta las tarifas en las constantes PARSE_CREDITS_PER_PAGE / EXTRACT_CREDITS_PER_PAGE de extract_facturas.py si cambian en tu cuenta o región.

Resultados de la prueba (tier cost_effective)

Probado con dos facturas de muestra públicas:

1. Factura de empresa con IVA 21% (factura-simple-iva21.pdf)

Campo	Valor extraído
Nº factura	98765
Fecha	24-11-2012
Emisor / NIF	DISTRIBUCIONES FICTICIAS… / X0000000T
Receptor / NIF	PRUEBASED GARCIA PROBERO / 77777777B
Líneas	2 (900,00 € + 126,00 €)
Base imponible	1026,00
IVA 21%	215,46
Total	1241,46
Forma de pago	CONTADO

2. Factura de suministro eléctrico — Iberdrola (factura-electricidad-comercio.pdf)

Campo	Valor extraído
Nº factura	21180613010076890
Fecha	13 de junio de 2018
Emisor / CIF	IBERDROLA CLIENTES, S.A.U. / A-95758389
Receptor / CIF	EXODO RENTAL S.L. / B86918002
Base imponible	66,90 (= 66,02 energía + 0,88 servicios)
IVA 21%	14,05
Total	80,95 EUR
Forma de pago	DOMICILIACION BANCARIA

Aciertos destacables: reconstruyó la base imponible sumando líneas, calculó la cuota de IVA, y dejó en null los campos que no aplican (IRPF en ambas, cantidad/precio unitario en la eléctrica) en lugar de inventarlos.

Facturas de ejemplo

invoices/ contiene un surtido de facturas reales de distintos emisores (suministros, SaaS, transporte, telecom…) para probar casos variados: IVA al 21%/10%, inversión del sujeto pasivo (IVA 0 en facturas intracomunitarias), datos anonimizados y documentos de varias páginas. Los datos personales de esas facturas no se documentan aquí ni deberían subirse a un repositorio público.

Paquete

Usa el SDK actual llama-cloud>=1.0 (from llama_cloud import LlamaCloud). El esquema Pydantic se pasa al servicio como JSON Schema vía FacturaEspanola.model_json_schema().

pyproject.toml

[project]
name = "extract-facturas-llama"
version = "0.1.0"
description = "Ejemplo de extracción de datos estructurados de facturas españolas con LlamaExtract"
requires-python = ">=3.10"
dependencies = [
    "llama-cloud>=1.0",
    "python-dotenv>=1.0",
    "pydantic>=2.0",
    "pypdf>=6.14.2",
]

schema.py

"""Esquema de extracción para una factura española válida.

El esquema es lo que guía al modelo: cada `Field(description=...)` actúa como una
instrucción para LlamaExtract sobre qué dato buscar en el documento. Por eso las
descripciones están redactadas con la terminología fiscal española (NIF/CIF, base
imponible, cuota de IVA, retención de IRPF...).
"""

from __future__ import annotations

from typing import List, Optional

from pydantic import BaseModel, Field, field_validator


class Parte(BaseModel):
    """Emisor o receptor de la factura."""

    nombre: str = Field(description="Nombre o razón social de la persona/empresa")
    nif: Optional[str] = Field(
        default=None,
        description="Identificación fiscal: NIF, CIF o DNI (p.ej. B12345678, 12345678Z)",
    )
    domicilio: Optional[str] = Field(
        default=None, description="Domicilio fiscal completo (calle, CP, localidad)"
    )


class LineaFactura(BaseModel):
    """Una línea o concepto facturado."""

    concepto: str = Field(description="Descripción del producto o servicio")
    cantidad: Optional[float] = Field(default=None, description="Unidades facturadas")
    precio_unitario: Optional[float] = Field(
        default=None, description="Precio por unidad, sin IVA"
    )
    tipo_iva: Optional[float] = Field(
        default=None, description="Tipo de IVA aplicado a la línea en %, p.ej. 21, 10, 4"
    )
    importe: Optional[float] = Field(
        default=None, description="Importe de la línea (cantidad x precio), sin IVA"
    )


class DesgloseIVA(BaseModel):
    """Una fila del desglose de IVA por tipo impositivo."""

    base_imponible: float = Field(description="Base imponible sujeta a este tipo de IVA")
    tipo_iva: float = Field(description="Tipo de IVA en %, p.ej. 21, 10, 4")
    cuota_iva: float = Field(description="Cuota de IVA resultante (base x tipo)")
    recargo_equivalencia: Optional[float] = Field(
        default=None, description="Cuota de recargo de equivalencia, si aplica"
    )


class FacturaEspanola(BaseModel):
    """Factura conforme al formato legal español."""

    numero_factura: str = Field(description="Número o serie-número de la factura")
    fecha_expedicion: str = Field(
        description="Fecha de expedición en formato original del documento"
    )

    emisor: Parte = Field(description="Quien emite la factura (vendedor)")
    receptor: Parte = Field(description="Quien recibe la factura (cliente)")

    lineas: List[LineaFactura] = Field(
        default_factory=list, description="Líneas de detalle / conceptos facturados"
    )
    desglose_iva: List[DesgloseIVA] = Field(
        default_factory=list,
        description="Desglose de IVA por tipo impositivo (puede haber varios tipos)",
    )

    base_imponible_total: Optional[float] = Field(
        default=None, description="Suma de todas las bases imponibles, sin IVA"
    )
    total_iva: Optional[float] = Field(
        default=None, description="Suma de todas las cuotas de IVA"
    )

    retencion_irpf_porcentaje: Optional[float] = Field(
        default=None, description="Porcentaje de retención de IRPF, típico en autónomos (p.ej. 15, 7)"
    )
    retencion_irpf_importe: Optional[float] = Field(
        default=None, description="Importe de la retención de IRPF (se resta del total)"
    )

    total_factura: float = Field(description="Importe total a pagar de la factura")
    moneda: Optional[str] = Field(default="EUR", description="Moneda, normalmente EUR")
    forma_pago: Optional[str] = Field(
        default=None, description="Forma o condiciones de pago (contado, transferencia...)"
    )

    @field_validator("lineas", "desglose_iva", mode="before")
    @classmethod
    def _none_to_empty_list(cls, v):
        """El extractor devuelve `null` (no la clave ausente) cuando una factura no
        tiene tabla de líneas o desglose. `default_factory` no cubre ese caso, así que
        convertimos `None` en lista vacía explícitamente."""
        return [] if v is None else v