Document Workflows

Document Workflows — ADE Pipeline Patterns

Overview

This skill provides reusable building blocks for composing LandingAI ADE primitives (parse, extract, split) into production-ready document processing pipelines. It complements the document-extraction skill:

Philosophy: Organize by workflow pattern (batch, RAG, DB insertion), not by document type. The same pattern applies whether documents are invoices, utility bills, or medical forms.

---

Step 0 (mandatory) — Pre-Flight Document Exploration {#pre-flight}

Run this before writing any pipeline code whenever working with documents whose internal structure has not already been inspected in this session.

**Rule: never write section-detection, heading-matching, or text-search code

without first running Tool 2 (diagnostic parse) on the sample document.

Heading format is document-specific and cannot be inferred from the task

description or document type alone — the only reliable way to know it is to

look at the actual ADE output.**

>

Common surprises: a paper's "Introduction" heading may appear as

1. Introduction (plain text, no #), ## Introduction, INTRODUCTION

(all-caps), or embedded inside a text chunk with body copy. Getting this

wrong means a silent failure (zero chunks matched) that requires a full

re-parse to debug.

Run Tool 1 (visual render) and Tool 2 (diagnostic parse) on 1–3 representative sample documents before writing any code. This takes under a minute and prevents debugging iterations that a pre-flight would have avoided.

Tool 1 — Visual page render

Render 1–2 pages as PNG and read them as visual context. No ADE credits used, but each PNG consumes context tokens. Use when layout is ambiguous or document origin is unknown (handwriting? scan? form?).

.venv/bin/python - << 'EOF'
import pymupdf
from pathlib import Path
from PIL import Image

pdf = Path('path/to/sample.pdf')
out_dir = Path('/tmp/ade_preflight'); out_dir.mkdir(exist_ok=True)
doc = pymupdf.open(pdf)
for pg in range(min(2, len(doc))):   # first 2 pages only
    pix = doc[pg].get_pixmap(matrix=pymupdf.Matrix(1.5, 1.5))   # 108 DPI
    img = Image.frombytes("RGB", [pix.width, pix.height], pix.samples)
    out = out_dir / f"{pdf.stem}_page{pg + 1}.png"
    img.save(out)
    print(out)
doc.close()
EOF

Then read the saved PNGs. Immediately answers:

• Are headings bold text (→ ADE may output plain-text heading, not # Heading)
• Is the document handwritten or scanned? → Tesseract OCR needed, not PyMuPDF
• Single-column or two-column layout?
• Any noise: running headers, page numbers, watermarks, stamps?

Tool 2 — ADE diagnostic parse

Parses 1 sample and prints markdown structure + chunk inventory. Uses ADE credits — keep to 1–3 samples only, never the full corpus.

.venv/bin/python - << 'EOF'
import os
from pathlib import Path
from collections import Counter
from dotenv import load_dotenv

# Load API key: prefer existing env var, then .env file lookup
load_dotenv()  # Load API key from .env. Add a path to the .env if needed.

from landingai_ade import LandingAIADE
client = LandingAIADE()
pr = client.parse(document=Path('path/to/sample.pdf'))

print("=== MARKDOWN (first 80 lines) ===")
for i, ln in enumerate(pr.markdown.splitlines()[:80], 1):
    print(f"{i:3}: {ln}")

print("\n=== CHUNKS ===")
for ch in pr.chunks:
    txt = (ch.markdown or '').replace('\n', ' ')[:70]
    b = ch.grounding.box
    print(f"p{ch.grounding.page} {ch.type:12} "
          f"l={b.left:.2f} t={b.top:.2f} r={b.right:.2f} b={b.bottom:.2f} | {txt}")

print(f"\nPages: {pr.metadata.page_count}  "
      f"Chunks: {len(pr.chunks)}  "
      f"Types: {dict(Counter(ch.type for ch in pr.chunks))}")
EOF

Cost note: Save the parse result with pr.model_dump() to a JSON file

after the first run. Load it for later development instead of calling

client.parse() again. Only re-parse when the document set changes.

What to look for

Tool 3 — Post-Crop Visual Verification (mandatory for bounding-box workflows) {#post-crop-verification}

After producing any bounding-box crop or overlay (figure extraction, chunk cropping, table cell extraction, word-level grounding), read back at least one output PNG as an image and describe what you see. Compare your description against the user's request. This catches:

• Wrong-page bugs — ADE page numbers are 0-indexed; an off-by-one error

lands the crop on an adjacent page with completely different content

• Wrong-region bugs — coordinate system mismatches that crop blank space

or an unrelated section

**Rule: never declare a crop workflow complete without visually reading at

least one output PNG and confirming its content matches the user's request.**

Verification steps

1. Save the first crop as PNG (the workflow already does this) 2. Read the PNG file as an image (use the read_file tool on the PNG path) 3. Describe what you see: what content, table, figure, or text appears? 4. Compare against the user's request:

• User asked for "the Events table" → does the crop show an Events table?
• User asked for "Figure 3" → does the crop show a chart/diagram?
• User asked for "Introduction section" → does the crop show intro text?

5. If the description doesn't match → investigate page indexing and bounding-box coordinates before continuing 6. Only proceed with remaining crops after the first one is verified

Why LLM vision, not heuristics

A blank-check heuristic (e.g. "mean brightness > 250 → blank") catches only the most obvious failures. The agent's own vision capability can semantically verify: "this crop shows a bar chart" vs "the user asked for a data table." This catches wrong-page errors even when the crop contains valid content from the wrong section.

---

Quick Reference — Building Blocks

---

Core Workflow: Parse + Extract + Save

The fundamental two-step ADE pattern. Every other workflow builds on this.

import io
from pathlib import Path
from typing import Any, Tuple, Type

from landingai_ade import LandingAIADE
from landingai_ade.lib import pydantic_to_json_schema


def parse_extract_save(
    doc_path: Path,
    client: LandingAIADE,
    schema_cls: Type[Any],
    output_dir: str = "./ade_results",
) -> Tuple[Any, Any]:
    """Parse a document, extract structured data, save both
    as JSON via save_to. Returns (parse_result, extract_result)."""
    # Step 1 — Parse (auto-saves {stem}_parse_output.json)
    parse_result = client.parse(
        document=doc_path, save_to=output_dir,
    )

    # Step 2 — Extract (auto-saves {stem}_extract_output.json)
    extract_result = client.extract(
        schema=pydantic_to_json_schema(schema_cls),
        markdown=io.BytesIO(
            parse_result.markdown.encode("utf-8")
        ),
        save_to=output_dir,
    )
    return parse_result, extract_result

`save_to` parameter: Available on parse(), extract(), and split().

Creates the folder if needed and writes {input_filename}_{method}_output.json.

This is a client-side convenience — the full response is saved locally after the API call.

Parse-Only (no extraction)

def parse_and_save(
    doc_path: Path,
    client: LandingAIADE,
    output_dir: str = "./ade_results",
) -> Any:
    return client.parse(
        document=doc_path, save_to=output_dir,
    )

Schemas: See schema-catalog.md for

ready-to-use Pydantic models (invoice, utility bill, bank statement,

pay stub, food label, CME certificate, document classifier).

See the document-extraction skill for schema design rules.

---

Classify-then-Extract

Process mixed document types by first classifying, then applying the appropriate schema. Two approaches:

Approach 1: Classification Extraction (any document mix)

from typing import Literal
from pydantic import BaseModel, Field


class DocType(BaseModel):
    type: Literal[
        "invoice", "bank_statement", "pay_stub",
        "utility_bill",
    ] = Field(description="The type of the document.")


# Map types to schemas (from schema-catalog.md)
SCHEMA_MAP: dict[str, type] = {
    "invoice": InvoiceSchema,
    "bank_statement": BankStatementSchema,
    "pay_stub": PayStubSchema,
    "utility_bill": UtilityBillSchema,
}


def classify_and_extract(
    doc_path: Path,
    client: LandingAIADE,
) -> dict:
    """Classify a document then extract with the matching
    schema."""
    pr = client.parse(document=doc_path)

    # Classify using first page
    cls = client.extract(
        schema=pydantic_to_json_schema(DocType),
        markdown=pr.markdown,
    )
    doc_type: str = cls.extraction["type"]

    # Extract with type-specific schema
    schema_cls = SCHEMA_MAP[doc_type]
    er = client.extract(
        schema=pydantic_to_json_schema(schema_cls),
        markdown=pr.markdown,
    )
    return {
        "type": doc_type,
        "extraction": er.extraction,
        "parse_result": pr,
        "extract_result": er,
    }

Approach 2: Split API (multi-document PDFs)

When a single PDF contains multiple document types (e.g., a packet with invoices + receipts), use the Split API first:

def split_classify_extract(
    pdf_path: Path,
    client: LandingAIADE,
    split_classes: list[dict],
) -> list[dict]:
    """Split a multi-doc PDF, classify each split, extract."""
    pr = client.parse(document=pdf_path, split="page")

    # Split into sub-documents
    split_result = client.split(
        markdown=pr.markdown,
        split_class=split_classes,
    )

    results = []
    for split_doc in split_result.splits:
        # Classify
        cls = client.extract(
            schema=pydantic_to_json_schema(DocType),
            markdown=split_doc.markdowns[0],
        )
        doc_type = cls.extraction["type"]

        # Extract
        schema_cls = SCHEMA_MAP[doc_type]
        er = client.extract(
            schema=pydantic_to_json_schema(schema_cls),
            markdown=split_doc.markdowns[0],
        )
        results.append({
            "type": doc_type,
            "extraction": er.extraction,
            "pages": split_doc.pages,
        })
    return results

Split API parameters: Use split_class (list of dicts with name, description, identifier keys).

See the document-extraction skill for full Split API reference.

When to use Split vs Classification:

- Split API: One PDF contains multiple separate documents

- Classification extraction: Each file is one document, but types vary

---

Section Extraction

Extract a named section (e.g. "Introduction", "Abstract") from a parsed document's markdown. Two approaches — choose based on document diversity and whether the extra API cost is justified.

Decision: If the diagnostic parse (Tool 2) shows consistent ATX headers (## Introduction, ## 2. Methods) across all your documents, use Approach A. If you see any plain-text numbered headings (1. Introduction) or formatting variation across documents, skip Approach A entirely and go straight to Approach B.

Approach A — Rule-based regex (free, fast, brittle)

ADE may emit headings as ATX markdown (## 2. Related Work) or plain-text (1. Introduction) even within the same document. Handle both patterns:

import re

def find_section(markdown: str, name: str) -> str | None:
    """Extract a named section from ADE markdown, handling both ATX
    headers (## Introduction) and plain-text numbered headings
    (1. Introduction) which ADE may emit inconsistently."""

    # Pattern 1: ATX header  (# Introduction, ## 1. Introduction …)
    m = re.search(
        r"^(#{1,6})\s+(?:\d+\.?\s+)?" + re.escape(name) + r"\b.*$",
        markdown, re.IGNORECASE | re.MULTILINE,
    )
    if m:
        level = len(m.group(1))
        end = re.search(r"^#{1," + str(level) + r"}\s",
                        markdown[m.end():], re.MULTILINE)
        end_pos = m.end() + (end.start() if end else len(markdown[m.end():]))
        return markdown[m.start():m.end() + end_pos].strip()

    # Pattern 2: plain-text numbered heading  (1. Introduction)
    m2 = re.search(r"^(?:\d+\.?\s+)?" + re.escape(name) + r"\s*$",
                   markdown, re.IGNORECASE | re.MULTILINE)
    if m2:
        end2 = re.search(
            r"^#{1,6}\s|^(?:\d+\.?\s+)[A-Z][a-zA-Z ]{3,}\s*$",
            markdown[m2.end():], re.MULTILINE,
        )
        end_pos = m2.end() + (end2.start() if end2 else len(markdown[m2.end():]))
        return markdown[m2.start():end_pos].strip()
    return None

Approach B — ADE extract (robust, handles document diversity)

Use ADE's own extraction to semantically locate sections — no regex needed. The LLM understands section meaning even when formatting is inconsistent:

from pydantic import BaseModel, Field
from landingai_ade import LandingAIADE
from landingai_ade.lib import pydantic_to_json_schema
from pathlib import Path


class PaperSections(BaseModel):
    abstract: str = Field(
        description="The abstract section, plain text only, "
                    "no markdown formatting or anchor tags."
    )
    introduction: str = Field(
        description="The introduction section, plain text only, "
                    "no markdown formatting or anchor tags."
    )


client = LandingAIADE()
pr = client.parse(document=Path("paper.pdf"))
er = client.extract(
    schema=pydantic_to_json_schema(PaperSections),
    markdown=pr.markdown,
)
intro_text = er.extraction["introduction"]

Cost note: Each extract() call consumes additional credits on top of

parse(). For high-volume pipelines with uniform document types, Approach A

avoids this cost. For diverse or unpredictable documents the accuracy

improvement justifies the extra credit.

---

Multi-Page Table Stitching {#table-stitching}

When a table spans multiple pages, ADE may emit it as separate table chunks per page — and may emit some pages as plain text instead of table chunks. This inconsistency can occur on any page, not just the last one.

Three approaches handle this, with different cost/accuracy/fragility trade-offs:

Decision guide:

• Use Approach A when accuracy is paramount and cost is secondary
• Use Approach B when rows are highly uniform, document structure is

predictable, and cost savings justify the fragility of regex-based parsing

• Use Approach C for quick prototyping or when missing some rows is

acceptable

Pre-flight additions for table stitching

Before choosing an approach, run the diagnostic parse (Tool 2) and check:

Domain-specific semantic checks

After stitching, add validation checks that leverage domain knowledge:

• Financial: running balances, column totals = sum of rows
• Inventory: quantity conservation across rows
• Time-series: chronological ordering, no sequence gaps
• Scientific: consistent units, monotonic IDs

These checks serve as both validation (confirming correctness) and disambiguation (resolving structural ambiguity in parsed output).

Full code for all three approaches with reusable patterns:

see table-stitching.md.

---

Batch Processing

Two patterns depending on scale. Both include per-document error handling.

Quick: ThreadPoolExecutor (sync)

from concurrent.futures import ThreadPoolExecutor, as_completed
from tqdm import tqdm


def batch_process(
    files: list[Path],
    schema_cls: type,
    max_workers: int = 4,
) -> list[tuple[Path, Any, Any]]:
    client = LandingAIADE()
    results: list[tuple[Path, Any, Any]] = []
    with ThreadPoolExecutor(max_workers=max_workers) as pool:
        futures = {
            pool.submit(
                parse_extract_save, fp, client, schema_cls
            ): fp
            for fp in files
        }
        for fut in tqdm(
            as_completed(futures), total=len(futures)
        ):
            fp = futures[fut]
            try:
                results.append((fp, *fut.result()))
            except Exception as e:
                print(f"FAILED {fp.name}: {e}")
    return results

Scalable: AsyncLandingAIADE (async)

import asyncio
from aiolimiter import AsyncLimiter
from landingai_ade import AsyncLandingAIADE


async def batch_parse_async(
    files: list[Path],
    rate_limit: int = 30,
) -> list[dict]:
    client = AsyncLandingAIADE()
    limiter = AsyncLimiter(rate_limit, 60)

    async def _process(fp: Path) -> dict | None:
        try:
            async with limiter:
                return {
                    "path": fp,
                    "result": await client.parse(document=fp),
                }
        except Exception as e:
            print(f"FAILED {fp.name}: {e}")
            return None

    raw = await asyncio.gather(*[_process(fp) for fp in files])
    return [r for r in raw if r]

Full code with output directory organization, CSV export, and chunk

image saving: see batch-processing.md.

---

Results to DataFrames and CSV

Flatten nested ADE extraction results into 4 normalized tables:

import uuid
from datetime import datetime, timezone


def rows_from_doc(
    file_path: str,
    parse_result: Any,
    extract_result: Any,
    run_id: str = "",
) -> tuple[dict, list[dict], list[dict], dict]:
    """Returns (main_row, line_rows, chunk_rows, md_record).

    - main_row: flattened top-level fields (nested__field)
    - line_rows: one per list item (line items, transactions)
    - chunk_rows: one per parsed chunk with bounding boxes
    - md_record: full markdown for traceability
    """
    doc_uuid = str(uuid.uuid4())
    f = extract_result.extraction

    # Flatten top-level fields
    main_row = {"doc_uuid": doc_uuid, "document_name": Path(file_path).name}
    for k, v in f.items():
        if isinstance(v, dict):
            for sk, sv in v.items():
                main_row[f"{k}__{sk}"] = sv
        elif not isinstance(v, list):
            main_row[k] = v

    # Extract list fields as line rows
    line_rows = [
        {"doc_uuid": doc_uuid, "list_field": k, "line_index": i, **item}
        for k, v in f.items() if isinstance(v, list)
        for i, item in enumerate(v) if isinstance(item, dict)
    ]

    # Chunk rows from parse result
    chunk_rows = [
        {
            "doc_uuid": doc_uuid,
            "chunk_id": getattr(ch, "id", None),
            "chunk_type": getattr(ch, "type", None),
            "page": ch.grounding.page if hasattr(ch, "grounding") else None,
        }
        for ch in (parse_result.chunks or [])
    ]

    md_record = {
        "doc_uuid": doc_uuid,
        "markdown": parse_result.markdown,
    }
    return main_row, line_rows, chunk_rows, md_record

Full code with Snowflake upload, UUID traceability, and bounding box

columns: see database-integration.md.

---

RAG Preparation

Quick path from parsed documents to a queryable RAG system. Two embedding options: local (free, offline) or API (higher quality).

Option A — Local embeddings with FastEmbed (free)

import re
from fastembed import TextEmbedding


def ade_to_embeddings_local(
    parse_results: list[dict],
    model: str = "BAAI/bge-small-en-v1.5",
) -> list[dict]:
    """Embed ADE chunks locally. Returns list of dicts with
    text, vector, and grounding metadata."""
    embedder = TextEmbedding(model_name=model)
    items: list[dict] = []
    for pr in parse_results:
        for ch in (pr["parse_result"].chunks or []):
            text = re.sub(
                r"<a id='[^']*'>\s*</a>", "", ch.markdown,
            ).strip()
            if not text:
                continue
            items.append({
                "text": text,
                "source": pr["name"],
                "page": ch.grounding.page,
                "box": {
                    "l": ch.grounding.box.left,
                    "t": ch.grounding.box.top,
                    "r": ch.grounding.box.right,
                    "b": ch.grounding.box.bottom,
                },
            })
    vecs = list(embedder.embed([i["text"] for i in items]))
    for item, vec in zip(items, vecs):
        item["vector"] = vec.tolist()
    return items

Option B — API embeddings with OpenAI

from langchain.docstore.document import Document
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings


def ade_to_rag(
    parse_results: list[dict],
    embedding_model: str = "text-embedding-3-small",
) -> FAISS:
    """Convert ADE parse results to a FAISS vector store.

    Args:
        parse_results: list of {"name": str, "parse_result": ParseResponse}
    """
    docs = [
        Document(
            page_content=ch.markdown,
            metadata={
                "source": item["name"],
                "chunk_type": getattr(ch, "type", ""),
                "page": ch.grounding.page if hasattr(ch, "grounding") else -1,
            },
        )
        for item in parse_results
        for ch in (item["parse_result"].chunks or [])
        if ch.markdown.strip()
    ]
    return FAISS.from_documents(
        docs, OpenAIEmbeddings(model=embedding_model)
    )

Full code with embedding best practices, hierarchical chunking,

multi-granularity strategies, ChromaDB, LangChain RetrievalQA, and

CSV export: see rag-pipelines.md.

Advanced RAG patterns in [rag-pipelines.md](references/rag-pipelines.md):

• Embedding computation (blocks 18–19) — choosing between local (FastEmbed, free) and API (OpenAI, higher quality) embeddings, including batch sizing and rate limiting
• Hierarchical chunking (block 20) — embed at multiple granularities (chunk, section, document) for hybrid retrieval
• Multi-granularity RAG (block 21) — combine chunk-level precision with document-level context, routing queries to the right embedding level based on scope

---

Visualization

Quick snippet for bounding box overlays on parsed pages:

from PIL import Image, ImageDraw
import pymupdf

CHUNK_COLORS = {
    "text": (40, 167, 69),
    "table": (0, 123, 255),
    "figure": (255, 0, 255),
    "marginalia": (111, 66, 193),
}

def annotate_page(
    img: Image.Image, chunks: list, page: int,
) -> Image.Image:
    annotated = img.copy()
    draw = ImageDraw.Draw(annotated)
    w, h = img.size
    for ch in chunks:
        if not hasattr(ch, "grounding") or ch.grounding.page != page:
            continue
        box = ch.grounding.box
        color = CHUNK_COLORS.get(getattr(ch, "type", ""), (200, 200, 200))
        draw.rectangle(
            [int(box.left * w), int(box.top * h),
             int(box.right * w), int(box.bottom * h)],
            outline=color, width=3,
        )
    return annotated

Full code with chunk image cropping, extraction-only overlays, and

word-level OCR grounding: see visualization.md.

---

Streamlit UI Pattern

Quick Streamlit app for interactive document processing:

import streamlit as st
from pathlib import Path
from landingai_ade import LandingAIADE
from landingai_ade.lib import pydantic_to_json_schema

st.title("Document Processor")

uploaded = st.file_uploader(
    "Upload document", type=["pdf", "png", "jpg"]
)
if uploaded:
    # Save temp file
    tmp = Path(f"/tmp/{uploaded.name}")
    tmp.write_bytes(uploaded.read())

    client = LandingAIADE()

    with st.spinner("Parsing..."):
        pr = client.parse(document=tmp)

    st.subheader("Markdown Preview")
    st.markdown(pr.markdown[:2000])

    st.subheader("Chunks")
    for ch in pr.chunks:
        with st.expander(
            f"{ch.type} (page {ch.grounding.page})"
        ):
            st.text(ch.markdown[:500])

<!-- Requires: pip install landingai-ade streamlit -->

Full Streamlit app with batch upload, extraction display, and

visualization tabs: adapt from the patterns in

batch-processing.md and

visualization.md.

---

Dependency Guide

---

Reference Files

Read these for full implementations when building a specific workflow:

• [schema-catalog.md](references/schema-catalog.md) — Ready-to-use Pydantic schemas for invoice, utility bill, bank statement, pay stub, food label, CME certificate, and document classification
• [batch-processing.md](references/batch-processing.md) — ThreadPoolExecutor, AsyncLandingAIADE, and Parse Jobs API patterns with full error handling
• [rag-pipelines.md](references/rag-pipelines.md) — Chunks to CSV, ChromaDB ingestion, FAISS + LangChain, and RAG query chains
• [database-integration.md](references/database-integration.md) — DataFrame normalization, Snowflake upload, and CSV export patterns
• [visualization.md](references/visualization.md) — Chunk image cropping, bounding box overlays, and word-level OCR grounding
• [table-stitching.md](references/table-stitching.md) — Parse+Extract (robust), HTML parsing (fragile), and pandas approaches for merging multi-page tables into a single output

Batch Processing Patterns

Three approaches for processing multiple documents, from simplest to most scalable. All patterns include per-document error handling so one failure doesn't stop the batch.

---

1. Sync Parallel — ThreadPoolExecutor

Best for: moderate batches (10–200 docs), simple scripts, notebooks.

import io
import json
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
from typing import Any, List, Tuple, Type

from landingai_ade import LandingAIADE
from landingai_ade.lib import pydantic_to_json_schema
from tqdm import tqdm


def parse_extract_save(
    doc_path: Path,
    client: LandingAIADE,
    schema_cls: Type[Any],
    output_dir: Path,
) -> Tuple[Any, Any]:
    """Parse one document, extract with schema, save both
    results as JSON. Returns (parse_result, extract_result)."""
    output_dir.mkdir(parents=True, exist_ok=True)
    stem = doc_path.stem

    # Step 1 — Parse
    parse_result = client.parse(document=doc_path)
    _save_json(
        parse_result, output_dir / f"parse_{stem}.json"
    )

    # Step 2 — Extract
    json_schema = pydantic_to_json_schema(schema_cls)
    extract_result = client.extract(
        schema=json_schema,
        markdown=io.BytesIO(
            parse_result.markdown.encode("utf-8")
        ),
    )
    _save_json(
        extract_result, output_dir / f"extract_{stem}.json"
    )
    return parse_result, extract_result


def _save_json(obj: Any, path: Path) -> None:
    data = (
        obj.model_dump()
        if hasattr(obj, "model_dump")
        else obj
    )
    path.write_text(
        json.dumps(data, indent=2, default=str),
        encoding="utf-8",
    )


def batch_parse_extract(
    file_paths: List[Path],
    schema_cls: Type[Any],
    output_dir: Path = Path("./ade_results"),
    max_workers: int = 4,
    api_key: str | None = None,
) -> List[Tuple[Path, Any, Any]]:
    """Process a list of documents in parallel.

    Returns list of (path, parse_result, extract_result)
    for successful documents. Failures are printed but
    do not stop the batch.
    """
    client = LandingAIADE(
        **({"apikey": api_key} if api_key else {})
    )
    results: List[Tuple[Path, Any, Any]] = []

    with ThreadPoolExecutor(max_workers=max_workers) as pool:
        futures = {
            pool.submit(
                parse_extract_save,
                fp, client, schema_cls, output_dir,
            ): fp
            for fp in file_paths
        }
        for future in tqdm(
            as_completed(futures),
            total=len(futures),
            desc="Processing",
        ):
            fp = futures[future]
            try:
                pr, er = future.result()
                results.append((fp, pr, er))
            except Exception as exc:
                print(f"FAILED {fp.name}: {exc}")

    return results

Usage

from pathlib import Path
from my_schema import InvoiceSchema  # or any Pydantic model

files = sorted(Path("invoices/").glob("*.pdf"))
results = batch_parse_extract(
    files,
    schema_cls=InvoiceSchema,
    output_dir=Path("./results"),
    max_workers=6,
)
print(f"Processed {len(results)}/{len(files)} documents")

---

2. Async Parallel — AsyncLandingAIADE

Best for: large batches (100+ docs), CLI tools, production pipelines. Uses asyncio + aiolimiter for rate-limited concurrency.

import asyncio
import json
from pathlib import Path
from typing import Any, Dict, List, Optional

import pandas as pd
from aiolimiter import AsyncLimiter
from landingai_ade import AsyncLandingAIADE


SUPPORTED_EXTS = {".pdf", ".png", ".jpg", ".jpeg"}


def collect_files(input_dir: Path) -> List[Path]:
    return sorted(
        p for p in input_dir.glob("*")
        if p.is_file() and p.suffix.lower() in SUPPORTED_EXTS
    )


async def process_document(
    file_path: Path,
    client: AsyncLandingAIADE,
    output_dirs: Dict[str, Path],
    rate_limiter: AsyncLimiter,
) -> Optional[Dict[str, Any]]:
    """Parse one document async, save JSON + markdown."""
    try:
        async with rate_limiter:
            result = await client.parse(document=file_path)

        stem = file_path.stem
        # Save JSON
        (output_dirs["json"] / f"{stem}.json").write_text(
            json.dumps(
                result.model_dump(), indent=2, default=str
            ),
            encoding="utf-8",
        )
        # Save markdown
        (output_dirs["markdown"] / f"{stem}.md").write_text(
            result.markdown, encoding="utf-8"
        )
        return {"path": file_path, "result": result}
    except Exception as exc:
        print(f"FAILED {file_path.name}: {exc}")
        return None


async def batch_parse_async(
    input_dir: Path,
    output_dir: Path,
    max_concurrent: int = 10,
    rate_limit: int = 30,
    api_key: str | None = None,
) -> List[Dict[str, Any]]:
    """Parse all documents in input_dir concurrently.

    Args:
        input_dir: folder with documents
        output_dir: base output folder (json/, markdown/
                    subdirs created automatically)
        max_concurrent: max parallel requests
        rate_limit: max requests per minute
    """
    files = collect_files(input_dir)
    if not files:
        print(f"No documents found in {input_dir}")
        return []

    # Create output subdirectories
    dirs: Dict[str, Path] = {}
    for sub in ("json", "markdown"):
        d = output_dir / sub
        d.mkdir(parents=True, exist_ok=True)
        dirs[sub] = d

    client = AsyncLandingAIADE(
        **({"apikey": api_key} if api_key else {})
    )
    limiter = AsyncLimiter(rate_limit, 60)

    tasks = [
        process_document(fp, client, dirs, limiter)
        for fp in files
    ]
    raw = await asyncio.gather(*tasks)
    return [r for r in raw if r is not None]

Usage

import asyncio
from pathlib import Path

results = asyncio.run(
    batch_parse_async(
        input_dir=Path("documents/"),
        output_dir=Path("results/"),
        max_concurrent=10,
        rate_limit=30,
    )
)
print(f"Parsed {len(results)} documents")

Adding Extraction to Async Pipeline

import io
from landingai_ade.lib import pydantic_to_json_schema


async def process_with_extraction(
    file_path: Path,
    client: AsyncLandingAIADE,
    schema_cls: type,
    output_dirs: Dict[str, Path],
    rate_limiter: AsyncLimiter,
) -> Optional[Dict[str, Any]]:
    try:
        async with rate_limiter:
            parse_result = await client.parse(
                document=file_path
            )
        async with rate_limiter:
            extract_result = await client.extract(
                schema=pydantic_to_json_schema(schema_cls),
                markdown=io.BytesIO(
                    parse_result.markdown.encode("utf-8")
                ),
            )
        return {
            "path": file_path,
            "parse": parse_result,
            "extract": extract_result,
        }
    except Exception as exc:
        print(f"FAILED {file_path.name}: {exc}")
        return None

---

3. Large File Processing — Parse Jobs API

Best for: files > 50 MB (up to ~1 GB). Uses async job submission + polling instead of synchronous upload.

import time
from pathlib import Path
from typing import Any

from landingai_ade import LandingAIADE


def parse_large_file(
    file_path: Path,
    client: LandingAIADE,
    poll_interval: int = 10,
    max_wait: int = 600,
) -> Any:
    """Submit a large file as a parse job and poll until
    complete.

    Returns the parse result (same shape as client.parse()).
    """
    # Step 1 — Submit job
    job = client.parse(
        document=file_path, is_async=True
    )
    job_id = job.request_id
    print(f"Job submitted: {job_id}")

    # Step 2 — Poll for completion
    elapsed = 0
    while elapsed < max_wait:
        status = client.get_parse_job(job_id)
        if status.status == "complete":
            print(f"Job complete after {elapsed}s")
            return status.data
        if status.status == "failed":
            raise RuntimeError(
                f"Parse job {job_id} failed: {status}"
            )
        time.sleep(poll_interval)
        elapsed += poll_interval

    raise TimeoutError(
        f"Job {job_id} not complete after {max_wait}s"
    )

Batch Large Files

from concurrent.futures import ThreadPoolExecutor

def batch_parse_large(
    file_paths: list[Path],
    max_workers: int = 3,
) -> list[Any]:
    client = LandingAIADE()
    results = []
    with ThreadPoolExecutor(max_workers=max_workers) as pool:
        futures = {
            pool.submit(
                parse_large_file, fp, client
            ): fp
            for fp in file_paths
        }
        for fut in futures:
            try:
                results.append(fut.result())
            except Exception as exc:
                fp = futures[fut]
                print(f"FAILED {fp.name}: {exc}")
    return results

---

Rate Limiting & Error Handling Tips

Dependencies

# Sync parallel (ThreadPoolExecutor)
pip install landingai-ade tqdm

# Async parallel
pip install landingai-ade aiolimiter pandas

# Large files — no extra deps beyond landingai-ade

Database Integration Patterns

Patterns for normalizing ADE extraction results into relational tables and loading them into databases. Covers DataFrame normalization, CSV export, and Snowflake insertion.

---

1. DataFrame Normalization

ADE extraction results are nested dicts. This pattern flattens them into 4 normalized tables suitable for any relational DB:

import uuid
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple


def _dig(obj: Any, *keys: str, default: Any = None) -> Any:
    """Safely traverse nested dicts/objects by key path."""
    for k in keys:
        if obj is None:
            return default
        if isinstance(obj, dict):
            obj = obj.get(k, default)
        else:
            obj = getattr(obj, k, default)
    return obj


def _to_float(v: Any) -> Optional[float]:
    if v is None:
        return None
    try:
        return float(v)
    except (ValueError, TypeError):
        return None


def rows_from_doc(
    file_path: str,
    parse_result: Any,
    extract_result: Any,
    run_id: str | None = None,
) -> Tuple[
    Dict[str, Any],
    List[Dict[str, Any]],
    List[Dict[str, Any]],
    Dict[str, Any],
]:
    """Transform ADE parse + extract results into 4 row types.

    Returns: (main_row, line_rows, chunk_rows, markdown_record)

    Args:
        file_path: original document path
        parse_result: from client.parse()
        extract_result: from client.extract()
        run_id: optional batch run identifier
    """
    doc_name = Path(file_path).name
    doc_uuid = str(uuid.uuid4())
    now = datetime.now(timezone.utc).isoformat()
    rid = run_id or doc_uuid

    f = extract_result.extraction  # dict
    m = getattr(extract_result, "extraction_metadata", {})

    # --- markdown record ---
    markdown_record = {
        "run_id": rid,
        "doc_uuid": doc_uuid,
        "document_name": doc_name,
        "processed_at": now,
        "markdown": parse_result.markdown,
    }

    # --- chunk rows ---
    chunk_rows: List[Dict[str, Any]] = []
    for ch in (parse_result.chunks or []):
        box = (
            ch.grounding.box
            if hasattr(ch, "grounding")
            and hasattr(ch.grounding, "box")
            else None
        )
        chunk_rows.append({
            "run_id": rid,
            "doc_uuid": doc_uuid,
            "document_name": doc_name,
            "chunk_id": getattr(ch, "id", None),
            "chunk_type": getattr(ch, "type", None),
            "text": getattr(ch, "markdown", None),
            "page": (
                ch.grounding.page
                if hasattr(ch, "grounding")
                else None
            ),
            "box_l": _to_float(box.left if box else None),
            "box_t": _to_float(box.top if box else None),
            "box_r": _to_float(box.right if box else None),
            "box_b": _to_float(box.bottom if box else None),
        })

    # --- main row (flatten top-level fields) ---
    main_row: Dict[str, Any] = {
        "run_id": rid,
        "doc_uuid": doc_uuid,
        "document_name": doc_name,
        "processed_at": now,
    }
    # Flatten one level of nesting
    for key, val in f.items():
        if isinstance(val, dict):
            for sub_key, sub_val in val.items():
                main_row[f"{key}__{sub_key}"] = sub_val
        elif isinstance(val, list):
            pass  # lists go to line_items
        else:
            main_row[key] = val

    # --- line item rows ---
    line_rows: List[Dict[str, Any]] = []
    for key, val in f.items():
        if not isinstance(val, list):
            continue
        for idx, item in enumerate(val):
            row: Dict[str, Any] = {
                "run_id": rid,
                "doc_uuid": doc_uuid,
                "document_name": doc_name,
                "list_field": key,
                "line_index": idx,
            }
            if isinstance(item, dict):
                row.update(item)
            else:
                row["value"] = item
            line_rows.append(row)

    return main_row, line_rows, chunk_rows, markdown_record

Usage — Build DataFrames from a Batch

import pandas as pd
from pathlib import Path
from landingai_ade import LandingAIADE
from landingai_ade.lib import pydantic_to_json_schema

client = LandingAIADE()
run_id = "batch_2025_01"

all_main, all_lines, all_chunks, all_md = [], [], [], []

for fp in Path("invoices/").glob("*.pdf"):
    pr = client.parse(document=fp)
    er = client.extract(
        schema=pydantic_to_json_schema(InvoiceSchema),
        markdown=pr.markdown,
    )
    main, lines, chunks, md = rows_from_doc(
        str(fp), pr, er, run_id=run_id
    )
    all_main.append(main)
    all_lines.extend(lines)
    all_chunks.extend(chunks)
    all_md.append(md)

df_main = pd.DataFrame(all_main)
df_lines = pd.DataFrame(all_lines)
df_chunks = pd.DataFrame(all_chunks)
df_md = pd.DataFrame(all_md)

# Save to CSV
for name, df in [
    ("main", df_main),
    ("line_items", df_lines),
    ("chunks", df_chunks),
    ("markdown", df_md),
]:
    df.to_csv(f"{run_id}_{name}.csv", index=False)

---

2. Snowflake Integration

Upload normalized tables to Snowflake using the connector's write_pandas or staged COPY pattern.

Note: ADE is also available as a Snowflake Native App (GA since Nov 2025), which runs ADE directly inside your Snowflake account without data leaving Snowflake. The patterns below use the standard Python SDK connector approach. For the Native App, see Snowflake Native App docs.

Connection Setup

import snowflake.connector
from snowflake.connector.pandas_tools import write_pandas


def get_snowflake_conn(
    account: str,
    user: str,
    password: str,
    database: str,
    schema: str,
    warehouse: str,
    role: str = "SYSADMIN",
) -> snowflake.connector.SnowflakeConnection:
    return snowflake.connector.connect(
        account=account,
        user=user,
        password=password,
        database=database,
        schema=schema,
        warehouse=warehouse,
        role=role,
    )

Table Creation

-- Main extraction results (one row per document)
CREATE TABLE IF NOT EXISTS ade_extractions (
    run_id          VARCHAR,
    doc_uuid        VARCHAR PRIMARY KEY,
    document_name   VARCHAR,
    processed_at    TIMESTAMP_TZ,
    -- Add flattened extraction columns here
    -- e.g., invoice_info__invoice_number VARCHAR
);

-- Line items (one row per repeating element)
CREATE TABLE IF NOT EXISTS ade_line_items (
    run_id          VARCHAR,
    doc_uuid        VARCHAR REFERENCES ade_extractions(doc_uuid),
    document_name   VARCHAR,
    list_field      VARCHAR,
    line_index      INTEGER,
    -- Add line item columns here
);

-- Parsed chunks with bounding boxes
CREATE TABLE IF NOT EXISTS ade_chunks (
    run_id          VARCHAR,
    doc_uuid        VARCHAR REFERENCES ade_extractions(doc_uuid),
    document_name   VARCHAR,
    chunk_id        VARCHAR,
    chunk_type      VARCHAR,
    text            VARCHAR,
    page            INTEGER,
    box_l           FLOAT,
    box_t           FLOAT,
    box_r           FLOAT,
    box_b           FLOAT
);

-- Full markdown for traceability
CREATE TABLE IF NOT EXISTS ade_markdown (
    run_id          VARCHAR,
    doc_uuid        VARCHAR REFERENCES ade_extractions(doc_uuid),
    document_name   VARCHAR,
    processed_at    TIMESTAMP_TZ,
    markdown        VARCHAR(16777216)
);

Upload DataFrames

def upload_to_snowflake(
    conn: snowflake.connector.SnowflakeConnection,
    df_main: "pd.DataFrame",
    df_lines: "pd.DataFrame",
    df_chunks: "pd.DataFrame",
    df_md: "pd.DataFrame",
) -> None:
    """Upload all 4 normalized tables to Snowflake."""
    # Column names must be UPPER CASE for Snowflake
    for table, df in [
        ("ADE_EXTRACTIONS", df_main),
        ("ADE_LINE_ITEMS", df_lines),
        ("ADE_CHUNKS", df_chunks),
        ("ADE_MARKDOWN", df_md),
    ]:
        if df.empty:
            continue
        df.columns = [c.upper() for c in df.columns]
        write_pandas(
            conn, df, table,
            auto_create_table=True,
            overwrite=False,
        )
        print(f"Uploaded {len(df)} rows to {table}")

Full Pipeline: Parse → Extract → Snowflake

import io
import pandas as pd
from pathlib import Path
from landingai_ade import LandingAIADE
from landingai_ade.lib import pydantic_to_json_schema


def ade_to_snowflake(
    input_dir: Path,
    schema_cls: type,
    sf_conn: "snowflake.connector.SnowflakeConnection",
    run_id: str = "default",
) -> int:
    """Parse, extract, normalize, and upload to Snowflake.

    Returns number of documents processed.
    """
    client = LandingAIADE()
    exts = {".pdf", ".png", ".jpg", ".jpeg"}
    files = [
        p for p in input_dir.glob("*")
        if p.suffix.lower() in exts
    ]

    all_main, all_lines, all_chunks, all_md = (
        [], [], [], []
    )
    for fp in files:
        try:
            pr = client.parse(document=fp)
            er = client.extract(
                schema=pydantic_to_json_schema(schema_cls),
                markdown=io.BytesIO(
                    pr.markdown.encode("utf-8")
                ),
            )
            main, lines, chunks, md = rows_from_doc(
                str(fp), pr, er, run_id=run_id
            )
            all_main.append(main)
            all_lines.extend(lines)
            all_chunks.extend(chunks)
            all_md.append(md)
        except Exception as exc:
            print(f"FAILED {fp.name}: {exc}")

    upload_to_snowflake(
        sf_conn,
        pd.DataFrame(all_main),
        pd.DataFrame(all_lines),
        pd.DataFrame(all_chunks),
        pd.DataFrame(all_md),
    )
    return len(all_main)

---

3. CSV Export Patterns

Summary CSV — One Row per Document

def extractions_to_summary_csv(
    results: list[tuple[str, dict]],
    output_path: Path,
) -> "pd.DataFrame":
    """Create a summary CSV with one row per document.

    Args:
        results: list of (filename, extraction_dict) tuples
        output_path: CSV file path
    """
    rows = []
    for name, extraction in results:
        row = {"document_name": name}
        for k, v in extraction.items():
            if isinstance(v, dict):
                for sk, sv in v.items():
                    row[f"{k}__{sk}"] = sv
            elif isinstance(v, list):
                row[f"{k}__count"] = len(v)
            else:
                row[k] = v
        rows.append(row)
    df = pd.DataFrame(rows)
    df.to_csv(output_path, index=False)
    return df

Per-Document JSON + Combined CSV

import json

def save_results(
    file_path: Path,
    parse_result: Any,
    extract_result: Any,
    output_dir: Path,
) -> None:
    """Save individual JSON files + append to combined CSV."""
    stem = file_path.stem
    output_dir.mkdir(parents=True, exist_ok=True)

    # Individual JSON
    for prefix, obj in [
        ("parse", parse_result),
        ("extract", extract_result),
    ]:
        data = (
            obj.model_dump()
            if hasattr(obj, "model_dump")
            else obj
        )
        (output_dir / f"{prefix}_{stem}.json").write_text(
            json.dumps(data, indent=2, default=str),
            encoding="utf-8",
        )

---

Dependencies

# DataFrame + CSV only
pip install landingai-ade pandas

# Snowflake integration
pip install landingai-ade pandas snowflake-connector-python[pandas]

# Environment variable management
pip install python-dotenv pydantic-settings

RAG Pipeline Patterns

End-to-end patterns for preparing ADE-parsed documents for Retrieval Augmented Generation (RAG) systems. Covers embedding computation, chunking strategies, vector DB ingestion, and query pipelines.

---

1. Chunks to CSV — RAG-Ready Dataset

Extract all chunks from parsed documents into a structured CSV with 19 columns including bounding boxes, sequence info, and metadata. This CSV can feed any vector DB or search index.

Grounding-aware records: Every record includes page, box_l,

box_t, box_r, box_b from ADE's grounding data. Preserve these

columns when ingesting into a vector DB — they let you trace retrieval

results back to exact document locations for highlighting or citation.

import re
from pathlib import Path
from typing import Any, Dict, List

import pandas as pd


def clean_chunk_text(text: str) -> str:
    """Remove anchor tags and strip whitespace."""
    cleaned = re.sub(r"<a id='[^']*'>\s*</a>", "", text)
    return cleaned.strip()


def chunks_to_records(
    parse_result: Any,
    document_name: str,
    model_version: str = "unknown",
) -> List[Dict[str, Any]]:
    """Convert parse result chunks to flat dicts.

    Each dict has 19 columns suitable for CSV / DataFrame:
    DOCUMENT_NAME, chunk_id, chunk_sequence_number,
    chunk_type, chunk_content_raw, chunk_content,
    chunk_text_length, chunk_word_count, page,
    box_l, box_t, box_r, box_b,
    prev_chunk_id, next_chunk_id, chunk_image_path,
    processed_at, ade_version, model_version
    """
    from datetime import datetime, timezone

    import landingai_ade

    chunks = parse_result.chunks or []
    now = datetime.now(timezone.utc).isoformat()
    records: List[Dict[str, Any]] = []

    for idx, ch in enumerate(chunks):
        raw = ch.markdown if hasattr(ch, "markdown") else ""
        clean = clean_chunk_text(raw)
        box = (
            ch.grounding.box
            if hasattr(ch, "grounding")
            and hasattr(ch.grounding, "box")
            else None
        )
        page = (
            ch.grounding.page
            if hasattr(ch, "grounding")
            else None
        )
        records.append({
            "DOCUMENT_NAME": document_name,
            "chunk_id": getattr(ch, "id", None),
            "chunk_sequence_number": idx,
            "chunk_type": getattr(ch, "type", None),
            "chunk_content_raw": raw,
            "chunk_content": clean,
            "chunk_text_length": len(clean),
            "chunk_word_count": len(clean.split()) if clean else 0,
            "page": page,
            "box_l": box.left if box else None,
            "box_t": box.top if box else None,
            "box_r": box.right if box else None,
            "box_b": box.bottom if box else None,
            "prev_chunk_id": (
                chunks[idx - 1].id if idx > 0 else None
            ),
            "next_chunk_id": (
                chunks[idx + 1].id
                if idx < len(chunks) - 1
                else None
            ),
            "chunk_image_path": None,
            "processed_at": now,
            "ade_version": landingai_ade.__version__,
            "model_version": model_version,
        })
    return records


def batch_chunks_to_csv(
    results: List[Dict[str, Any]],
    output_path: Path,
) -> pd.DataFrame:
    """Combine chunk records from multiple documents into
    one CSV.

    Args:
        results: list of dicts with keys 'name' and
                 'parse_result'
        output_path: CSV file path
    """
    all_records: List[Dict[str, Any]] = []
    for r in results:
        all_records.extend(
            chunks_to_records(r["parse_result"], r["name"])
        )
    df = pd.DataFrame(all_records)
    df.to_csv(output_path, index=False)
    return df

Usage

from landingai_ade import LandingAIADE
from pathlib import Path

client = LandingAIADE()
results = []
for fp in Path("docs/").glob("*.pdf"):
    pr = client.parse(document=fp)
    results.append({"name": fp.name, "parse_result": pr})

df = batch_chunks_to_csv(results, Path("all_chunks.csv"))
print(f"{len(df)} chunks from {df['DOCUMENT_NAME'].nunique()} docs")

---

2. Vector DB Ingestion — ChromaDB

Local persistent vector store using OpenAI embeddings. Good for prototyping and small-to-medium corpora.

from pathlib import Path
from typing import Any, List

import chromadb
from chromadb.config import Settings
from openai import OpenAI


def ade_chunks_to_chromadb(
    parse_results: List[dict],
    collection_name: str = "ade_documents",
    persist_dir: str = "./chroma_db",
    embedding_model: str = "text-embedding-3-small",
) -> chromadb.Collection:
    """Ingest ADE chunks into a persistent ChromaDB collection.

    Args:
        parse_results: list of dicts with 'name' (str) and
                       'parse_result' (ParseResponse)
        collection_name: ChromaDB collection name
        persist_dir: directory for persistent storage
        embedding_model: OpenAI embedding model name

    Returns:
        The ChromaDB collection with all chunks ingested.
    """
    openai_client = OpenAI()
    chroma_client = chromadb.PersistentClient(
        path=persist_dir
    )
    collection = chroma_client.get_or_create_collection(
        name=collection_name,
        metadata={"hnsw:space": "cosine"},
    )

    for doc in parse_results:
        name = doc["name"]
        chunks = doc["parse_result"].chunks or []

        texts, ids, metadatas = [], [], []
        for ch in chunks:
            text = ch.markdown if hasattr(ch, "markdown") else ""
            if not text.strip():
                continue
            chunk_id = f"{name}:{ch.id}"
            texts.append(text)
            ids.append(chunk_id)
            metadatas.append({
                "document": name,
                "chunk_type": getattr(ch, "type", "unknown"),
                "page": (
                    ch.grounding.page
                    if hasattr(ch, "grounding")
                    else -1
                ),
            })

        if not texts:
            continue

        # Generate embeddings in batches of 100
        all_embeddings: List[List[float]] = []
        for i in range(0, len(texts), 100):
            batch = texts[i : i + 100]
            resp = openai_client.embeddings.create(
                input=batch, model=embedding_model
            )
            all_embeddings.extend(
                [e.embedding for e in resp.data]
            )

        collection.add(
            ids=ids,
            documents=texts,
            embeddings=all_embeddings,
            metadatas=metadatas,
        )

    return collection

Query ChromaDB

def query_chromadb(
    collection: chromadb.Collection,
    question: str,
    n_results: int = 5,
    embedding_model: str = "text-embedding-3-small",
) -> dict:
    """Query the collection and return matching chunks."""
    openai_client = OpenAI()
    resp = openai_client.embeddings.create(
        input=[question], model=embedding_model
    )
    query_embedding = resp.data[0].embedding
    return collection.query(
        query_embeddings=[query_embedding],
        n_results=n_results,
    )

---

3. Vector DB Ingestion — FAISS + LangChain

For LangChain-based RAG pipelines. Uses FAISS for in-memory vector search.

from typing import Any, List

from langchain.docstore.document import Document
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings


def ade_to_langchain_docs(
    parse_results: List[dict],
) -> List[Document]:
    """Convert ADE parse results to LangChain Documents.

    Each chunk becomes one Document with metadata including
    source document name, chunk type, and page number.
    """
    docs: List[Document] = []
    for item in parse_results:
        name = item["name"]
        chunks = item["parse_result"].chunks or []
        for ch in chunks:
            text = ch.markdown if hasattr(ch, "markdown") else ""
            if not text.strip():
                continue
            docs.append(Document(
                page_content=text,
                metadata={
                    "source": name,
                    "chunk_type": getattr(ch, "type", "unknown"),
                    "chunk_id": getattr(ch, "id", ""),
                    "page": (
                        ch.grounding.page
                        if hasattr(ch, "grounding")
                        else -1
                    ),
                },
            ))
    return docs


def build_faiss_index(
    documents: List[Document],
    embedding_model: str = "text-embedding-3-small",
) -> FAISS:
    """Build a FAISS vector store from LangChain Documents."""
    embeddings = OpenAIEmbeddings(model=embedding_model)
    return FAISS.from_documents(documents, embeddings)

RAG Query with LangChain

from langchain.chains import RetrievalQA
from langchain_openai import ChatOpenAI


def build_rag_chain(
    vectorstore: FAISS,
    model: str = "gpt-4o-mini",
    k: int = 5,
) -> RetrievalQA:
    """Build a RetrievalQA chain from a FAISS index."""
    retriever = vectorstore.as_retriever(
        search_kwargs={"k": k}
    )
    return RetrievalQA.from_chain_type(
        llm=ChatOpenAI(model=model, temperature=0),
        chain_type="stuff",
        retriever=retriever,
        return_source_documents=True,
    )


# Usage
chain = build_rag_chain(vectorstore)
answer = chain.invoke({"query": "What is the total revenue?"})
print(answer["result"])
for doc in answer["source_documents"]:
    print(f"  - {doc.metadata['source']} p{doc.metadata['page']}")

---

4. Full RAG Pipeline — End to End

Combines parsing, chunking, vector DB, and querying into one flow.

import asyncio
from pathlib import Path

from langchain.chains import RetrievalQA
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from landingai_ade import LandingAIADE


def build_rag_from_folder(
    input_dir: Path,
    embedding_model: str = "text-embedding-3-small",
    llm_model: str = "gpt-4o-mini",
) -> RetrievalQA:
    """One-shot: parse all docs in a folder and build a
    RAG chain ready for querying.

    Returns a LangChain RetrievalQA chain.
    """
    client = LandingAIADE()
    exts = {".pdf", ".png", ".jpg", ".jpeg"}
    files = [
        p for p in input_dir.glob("*")
        if p.suffix.lower() in exts
    ]

    # Parse all documents
    parse_results = []
    for fp in files:
        pr = client.parse(document=fp)
        parse_results.append({"name": fp.name, "parse_result": pr})

    # Convert to LangChain docs
    docs = ade_to_langchain_docs(parse_results)

    # Build vector store
    embeddings = OpenAIEmbeddings(model=embedding_model)
    vectorstore = FAISS.from_documents(docs, embeddings)

    # Build chain
    return RetrievalQA.from_chain_type(
        llm=ChatOpenAI(model=llm_model, temperature=0),
        chain_type="stuff",
        retriever=vectorstore.as_retriever(
            search_kwargs={"k": 5}
        ),
        return_source_documents=True,
    )

Usage

chain = build_rag_from_folder(Path("10k_filings/"))
result = chain.invoke(
    {"query": "What were the main risk factors?"}
)
print(result["result"])

---

5. Embedding Computation

Two approaches for computing embeddings from ADE chunks: local (free, offline, fast) and API-based (higher quality, paid). Choose based on your cost/quality tradeoff.

Local Embeddings with FastEmbed

Uses FastEmbed to run embedding models locally. No API key needed, no per-token cost, works offline.

from typing import Any

from fastembed import TextEmbedding


def compute_embeddings_local(
    texts: list[str],
    model: str = "BAAI/bge-small-en-v1.5",
) -> list[list[float]]:
    """Embed texts locally with FastEmbed (batched).

    Default model: BAAI/bge-small-en-v1.5 (384 dims, ~33M params).
    Other options:
      - BAAI/bge-base-en-v1.5  (768 dims, ~110M params)
      - sentence-transformers/all-MiniLM-L6-v2  (384 dims)
    """
    embedder = TextEmbedding(model_name=model)
    return [v.tolist() for v in embedder.embed(texts)]

API Embeddings with OpenAI

Higher quality, especially for domain-specific content. Requires OPENAI_API_KEY and incurs per-token cost.

from openai import OpenAI


def compute_embeddings_openai(
    texts: list[str],
    model: str = "text-embedding-3-small",
    batch_size: int = 100,
) -> list[list[float]]:
    """Embed texts via OpenAI API in batches."""
    client = OpenAI()
    all_vecs: list[list[float]] = []
    for i in range(0, len(texts), batch_size):
        resp = client.embeddings.create(
            input=texts[i : i + batch_size], model=model,
        )
        all_vecs.extend(e.embedding for e in resp.data)
    return all_vecs

Embedding Best Practices

Self-Describing Embedding Output

Always store the embedding model name and dimensions alongside the vector so downstream consumers can interpret it correctly:

def make_embedding_record(
    text: str,
    vector: list[float],
    model: str,
    metadata: dict | None = None,
) -> dict:
    """Wrap a vector with its model info and metadata."""
    return {
        "text": text,
        "embedding": {
            "model": model,
            "dimensions": len(vector),
            "vector": vector,
        },
        "metadata": metadata or {},
    }

Model Selection Guide

---

6. Multi-Granularity Embedding Strategy

ADE chunks are the finest-grained unit, but they're not always the right unit for embedding. Choose the granularity that matches your retrieval needs.

Granularity Levels

Chunk-Level (default)

Each ADE chunk gets its own embedding. This is what Sections 2–4 above use. Fine-grained but may split semantic units across multiple vectors.

# Already shown in Sections 2-4 — each chunk → one embedding
texts = [
    clean_chunk_text(ch.markdown)
    for ch in parse_result.chunks
    if ch.markdown.strip()
    and getattr(ch, "type", "") in {"text", "table", "card"}
]

Hierarchical Chunking

Group consecutive ADE chunks into higher-level semantic units before embedding. The grouping boundary is document-specific — the pattern is always the same but the boundary detection varies:

• Heading detection (regex or ADE extract) — for papers, reports
• Clause boundaries (ADE split API or extract) — for contracts
• Page boundaries — simple, works for any document
• Fixed-size sliding windows — N consecutive chunks with overlap

The abstract pattern:

from dataclasses import dataclass, field
from typing import Any, Callable


@dataclass
class ChunkGroup:
    """A group of consecutive ADE chunks forming a semantic unit."""
    label: str
    chunks: list[Any] = field(default_factory=list)
    grounding_refs: list[dict] = field(default_factory=list)

    @property
    def text(self) -> str:
        return "\n".join(
            clean_chunk_text(ch.markdown)
            for ch in self.chunks if ch.markdown.strip()
        )

    @property
    def embedding_input(self) -> str:
        """Prepend label for better embedding quality."""
        return f"{self.label}\n\n{self.text}"


def group_chunks(
    chunks: list[Any],
    is_boundary: Callable[[Any], str | None],
) -> list[ChunkGroup]:
    """Group ADE chunks by a boundary detection function.

    Args:
        chunks: ADE parse_result.chunks
        is_boundary: function that returns a group label
            (str) if the chunk starts a new group, or
            None if it continues the current group.

    Returns:
        List of ChunkGroup with grounding refs preserved.
    """
    groups: list[ChunkGroup] = []
    current: ChunkGroup | None = None
    for ch in chunks:
        label = is_boundary(ch)
        if label is not None:
            current = ChunkGroup(label=label)
            groups.append(current)
        if current is None:
            current = ChunkGroup(label="(preamble)")
            groups.append(current)
        current.chunks.append(ch)
        if hasattr(ch, "grounding"):
            b = ch.grounding.box
            current.grounding_refs.append({
                "page": ch.grounding.page,
                "box": {
                    "left": b.left, "top": b.top,
                    "right": b.right, "bottom": b.bottom,
                },
            })
    return groups

Example boundary detectors:

import re

# Page-based: new group every page
def by_page(ch: Any) -> str | None:
    page = ch.grounding.page if hasattr(ch, "grounding") else -1
    return f"Page {page + 1}" if not hasattr(by_page, "_last") or by_page._last != page else None
    # (simplified — use a closure or class for production)

# Heading-based: new group on ATX or numbered headings
def by_heading(ch: Any) -> str | None:
    text = re.sub(r"<a id='[^']*'></a>\s*", "", ch.markdown or "").strip()
    first_line = text.split("\n")[0].strip()
    if re.match(r"^#{1,6}\s+", first_line):
        return re.sub(r"^#{1,6}\s+", "", first_line)
    if re.match(r"^\d+(?:\.\d+)*\.?\s+[A-Z]", first_line):
        return first_line
    return None

Using groups for embedding:

groups = group_chunks(parse_result.chunks, by_heading)
texts = [g.embedding_input for g in groups if g.text.strip()]
vectors = compute_embeddings_local(texts)

# Each group carries grounding_refs for source attribution
for g, vec in zip(groups, vectors):
    print(f"{g.label}: {len(g.grounding_refs)} chunk refs, "
          f"{len(vec)} dims")

Document-Level

Embed the full document markdown or a summary. Useful for routing queries to the right document before doing fine-grained search.

# Full markdown (may be long — consider truncation)
doc_text = parse_result.markdown[:8000]
doc_vec = compute_embeddings_local([doc_text])[0]

# Or use ADE extract to get a summary first
from landingai_ade.lib import pydantic_to_json_schema
from pydantic import BaseModel, Field

class DocSummary(BaseModel):
    summary: str = Field(
        description="A 2-3 sentence summary of the document."
    )

er = client.extract(
    schema=pydantic_to_json_schema(DocSummary),
    markdown=parse_result.markdown,
)
summary_vec = compute_embeddings_local(
    [er.extraction["summary"]]
)[0]

Decision Matrix

---

Chunk Filtering Tips

Not all chunks are useful for RAG. Filter by type to improve relevance:

# Keep only text and table chunks (skip logos, scan codes)
RAG_CHUNK_TYPES = {"text", "table", "card"}

docs = [
    Document(page_content=ch.markdown, metadata={...})
    for ch in parse_result.chunks
    if getattr(ch, "type", "") in RAG_CHUNK_TYPES
    and ch.markdown.strip()
]

---

Dependencies

# Chunks to CSV only
pip install landingai-ade pandas

# Local embeddings (free, offline)
pip install landingai-ade fastembed

# ChromaDB pipeline (API embeddings)
pip install landingai-ade chromadb openai

# FAISS + LangChain pipeline (API embeddings)
pip install landingai-ade langchain langchain-openai langchain-community faiss-cpu

# Local embeddings + ChromaDB
pip install landingai-ade fastembed chromadb

# Full pipeline (all options)
pip install landingai-ade pandas fastembed chromadb openai langchain langchain-openai langchain-community faiss-cpu

Schema Catalog — Ready-to-Use Pydantic Models

Ready-to-use extraction schemas for common document types. Each schema is a Pydantic BaseModel that can be converted to JSON Schema via pydantic_to_json_schema and passed to client.extract().

Tip: ADE supports one level of nesting. Use nested BaseModel

sub-classes for logical grouping, and List[SubModel] for repeating items

like line items or transactions.

---

Usage Pattern (all schemas)

from landingai_ade import LandingAIADE
from landingai_ade.lib import pydantic_to_json_schema

client = LandingAIADE()
parse_result = client.parse(document=path)
extract_result = client.extract(
    schema=pydantic_to_json_schema(MySchema),
    markdown=parse_result.markdown,
)
data: dict = extract_result.extraction

---

1. Invoice Schema

6 nested groups, 30+ fields. Covers invoices from any vendor/country.

from typing import Optional, List
from datetime import date
from pydantic import BaseModel, Field


class DocumentInfo(BaseModel):
    invoice_date_raw: str = Field(
        ...,
        description=(
            "Invoice date as found in the document."
            " Do not reformat."
        ),
    )
    invoice_date: Optional[date] = Field(
        ..., description="Invoice date in YYYY-MM-DD."
    )
    invoice_number: str = Field(
        ..., description="Invoice number."
    )
    order_date: Optional[str] = Field(
        None, description="Order or purchase date."
    )
    po_number: Optional[str] = Field(
        None, description="Customer purchase order (PO) number."
    )
    status: Optional[str] = Field(
        None,
        description="Payment status (e.g., PAID, UNPAID).",
    )


class CustomerInfo(BaseModel):
    sold_to_name: str = Field(
        ...,
        description=(
            "Name of the customer billed."
            " Can be a person or an organization."
        ),
    )
    sold_to_address: Optional[str] = Field(
        None, description="Address of the customer billed."
    )
    customer_email: Optional[str] = Field(
        None, description="Email address for the customer."
    )


class SupplierInfo(BaseModel):
    supplier_name: str = Field(
        ..., description="Name of the supplier company."
    )
    supplier_address: Optional[str] = Field(
        None, description="Address of the supplier."
    )
    representative: Optional[str] = Field(
        None, description="Sales representative(s)."
    )
    email: Optional[str] = Field(
        None, description="Email address of the supplier."
    )
    phone: Optional[str] = Field(
        None, description="Phone number of the supplier."
    )
    gstin: Optional[str] = Field(
        None, description="GSTIN of the supplier (India)."
    )
    pan: Optional[str] = Field(
        None, description="Permanent Account Number (India)."
    )


class TermsAndShipping(BaseModel):
    payment_terms: Optional[str] = Field(
        None, description="Payment terms (e.g., Net 30)."
    )
    ship_via: Optional[str] = Field(
        None, description="Carrier/service (e.g., UPS Ground)."
    )
    ship_date: Optional[str] = Field(
        None, description="Date shipped."
    )
    tracking_number: Optional[str] = Field(
        None, description="Tracking number."
    )


class TotalsSummary(BaseModel):
    currency: Optional[str] = Field(
        None, description="ISO currency code."
    )
    total_due_raw: Optional[str] = Field(
        None, description="Total due as shown in the doc."
    )
    total_due: float = Field(
        ..., description="Total amount due (numeric, no symbols)."
    )
    subtotal: Optional[float] = Field(
        None, description="Subtotal (numeric)."
    )
    tax: Optional[float] = Field(
        None, description="Tax (numeric)."
    )
    shipping: Optional[float] = Field(
        None, description="Shipping (numeric)."
    )
    handling_fee: Optional[float] = Field(
        None, description="Handling fee (numeric)."
    )


class LineItem(BaseModel):
    line_number: Optional[str] = Field(
        None, description="Printed line number."
    )
    sku: Optional[str] = Field(
        None, description="SKU / Item code / Part number."
    )
    description: str = Field(
        ..., description="Item or service description."
    )
    quantity: Optional[float] = Field(
        None, description="Quantity purchased."
    )
    unit_price: Optional[float] = Field(
        None, description="Unit price (numeric)."
    )
    amount: Optional[float] = Field(
        None, description="Extended line amount (numeric)."
    )


class InvoiceSchema(BaseModel):
    invoice_info: DocumentInfo = Field(
        description="Key identifiers and dates."
    )
    customer_info: CustomerInfo = Field(
        description="Details about the customer billed."
    )
    company_info: SupplierInfo = Field(
        description="Details about the issuing company."
    )
    order_details: TermsAndShipping = Field(
        description="Payment and shipping information."
    )
    totals_summary: TotalsSummary = Field(
        description="Financial totals by category."
    )
    line_items: List[LineItem] = Field(
        default_factory=list,
        description="List of items included in the invoice.",
    )

---

2. Utility Bill Schema

Provider, account, billing summary, electric and gas charges.

from typing import Optional
from pydantic import BaseModel, Field


class ProviderInfo(BaseModel):
    provider: str = Field(
        ..., description="Name of the utility provider."
    )
    phone_number: Optional[str] = Field(
        None,
        description="Customer service phone (XXX-XXX-XXXX).",
    )
    website: Optional[str] = Field(
        None, description="Official website URL."
    )
    usage_bar_chart: bool = Field(
        False,
        description="Does the bill include a usage trend chart?",
    )


class AccountInfo(BaseModel):
    account_holder: str = Field(
        ...,
        description=(
            "Full name of the account holder."
            " May be a person or organization."
        ),
    )
    account_number: str = Field(
        ..., description="Unique customer account identifier."
    )
    service_address: str = Field(
        ...,
        description=(
            "Full service address on one line"
            " (remove newlines, replace with space)."
        ),
    )
    service_address_city: Optional[str] = Field(
        None, description="City of service address."
    )
    service_address_state: Optional[str] = Field(
        None, description="2-letter state abbreviation."
    )
    service_address_zip: Optional[str] = Field(
        None, description="5-digit ZIP code."
    )


class BillingSummary(BaseModel):
    due_date: str = Field(
        ..., description="Payment due date (YYYY-MM-DD)."
    )
    bill_date: str = Field(
        ..., description="Bill issue date (YYYY-MM-DD)."
    )
    service_start_date: Optional[str] = Field(
        None, description="Service period start (MM-DD-YYYY)."
    )
    service_end_date: Optional[str] = Field(
        None, description="Service period end (MM-DD-YYYY)."
    )
    total_amount_due: str = Field(
        ...,
        description="Total amount due including currency symbol.",
    )


class ElectricCharges(BaseModel):
    meter_number: Optional[str] = Field(
        None,
        description=(
            "Electric meter identifier."
            " Blank if no electric service."
        ),
    )
    usage_kwh: Optional[str] = Field(
        None, description="Total kWh for billing period."
    )
    total_electric_charges: Optional[str] = Field(
        None,
        description="Total electric charges with currency symbol.",
    )


class GasCharges(BaseModel):
    meter_number: Optional[str] = Field(
        None,
        description=(
            "Gas meter identifier."
            " Blank if no gas service."
        ),
    )
    usage_therms: Optional[str] = Field(
        None, description="Total therms for billing period."
    )
    total_gas_charges: Optional[str] = Field(
        None,
        description="Total gas charges with currency symbol.",
    )


class UtilityBillSchema(BaseModel):
    provider_info: ProviderInfo = Field(
        description="Energy provider details."
    )
    account_info: AccountInfo = Field(
        description="Account and customer identifiers."
    )
    billing_summary: BillingSummary = Field(
        description="Charges and due dates."
    )
    electric_charges: ElectricCharges = Field(
        description="Electric usage and charges."
    )
    gas_charges: GasCharges = Field(
        description="Gas usage and charges."
    )

---

3. Bank Statement Schema

from typing import Optional, List
from pydantic import BaseModel, Field


class BankTransaction(BaseModel):
    date: str = Field(
        ..., description="Transaction date (YYYY-MM-DD)."
    )
    description: str = Field(
        ..., description="Transaction description."
    )
    amount: float = Field(
        ..., description="Transaction amount (numeric)."
    )
    type: Optional[str] = Field(
        None,
        description="Transaction type: debit or credit.",
    )


class BankStatementSchema(BaseModel):
    bank_name: str = Field(
        ..., description="Name of the bank."
    )
    account_number: str = Field(
        ..., description="Bank account number."
    )
    statement_period: Optional[str] = Field(
        None,
        description="Statement period (e.g., Jan 1 - Jan 31, 2025).",
    )
    opening_balance: Optional[float] = Field(
        None, description="Opening balance (numeric)."
    )
    closing_balance: float = Field(
        ..., description="Closing / current balance (numeric)."
    )
    total_deposits: Optional[float] = Field(
        None, description="Total deposits (numeric)."
    )
    total_withdrawals: Optional[float] = Field(
        None, description="Total withdrawals (numeric)."
    )
    transactions: List[BankTransaction] = Field(
        default_factory=list,
        description="List of transactions in the statement.",
    )

---

3b. Multi-Page Table Schema (for Table Stitching)

Use this pattern when a table spans multiple pages and you want the LLM to stitch all rows into a single list via client.extract().

from typing import List, Optional
from pydantic import BaseModel, Field


class TableRow(BaseModel):
    """Generic row — customize fields for your table."""
    key_column: str = Field(
        description=(
            "Primary identifier (e.g., date, ID). "
            "Use empty string for continuation rows."
        )
    )
    description: str = Field(
        description="Description or label column."
    )
    amount_a: Optional[str] = Field(
        default=None,
        description=(
            "First amount column (digits and commas only, "
            "no currency symbol). "
            "Null if not applicable for this row."
        ),
    )
    amount_b: Optional[str] = Field(
        default=None,
        description=(
            "Second amount column. "
            "Null if not applicable."
        ),
    )
    running_total: str = Field(
        description=(
            "Running total or balance after this row."
        )
    )


class MultiPageTable(BaseModel):
    rows: List[TableRow] = Field(
        description=(
            "All data rows in order across ALL pages of "
            "the document. Include rows from every page, "
            "even if some pages render as plain text "
            "rather than tables. "
            "Skip column-header rows and section-header "
            "rows."
        )
    )

Schema design tips for multi-page tables:

• Say "across ALL pages" in the List field description
• Mention "even if some pages render as plain text"
• Say "Skip column-header rows" to avoid duplicated headers
• Use Optional[str] for mutually exclusive amount columns
• Use str (not float) for amounts to preserve original formatting
• Add domain-specific hints in descriptions (e.g., "running balance

after this transaction") to help the LLM resolve ambiguities

Full patterns for three table stitching approaches (parse+extract,

HTML parsing, pandas): see

table-stitching.md.

---

4. Pay Stub Schema

from typing import Optional, List
from pydantic import BaseModel, Field


class Deduction(BaseModel):
    name: str = Field(
        ..., description="Deduction name (e.g., Federal Tax)."
    )
    amount: float = Field(
        ..., description="Deduction amount (numeric)."
    )


class PayStubSchema(BaseModel):
    employee_name: str = Field(
        ..., description="Full name of the employee."
    )
    employer_name: Optional[str] = Field(
        None, description="Name of the employer."
    )
    pay_period: str = Field(
        ..., description="Pay period covered by this stub."
    )
    pay_date: Optional[str] = Field(
        None, description="Payment date (YYYY-MM-DD)."
    )
    gross_pay: float = Field(
        ..., description="Gross pay amount (numeric)."
    )
    net_pay: float = Field(
        ..., description="Net pay after deductions (numeric)."
    )
    total_deductions: Optional[float] = Field(
        None, description="Total deductions (numeric)."
    )
    deductions: List[Deduction] = Field(
        default_factory=list,
        description="Itemized deductions.",
    )

---

5. Food / Product Label Schema

27 fields covering identification, weight, certifications, and dietary claims.

from pydantic import BaseModel, Field


class ProductLabelSchema(BaseModel):
    # Identification
    product_name: str = Field(
        ...,
        description=(
            "Full product name excluding brand"
            " as it appears on packaging."
        ),
    )
    brand: str = Field(
        ..., description="Brand or company name."
    )
    product_type: str = Field(
        ...,
        description=(
            "General category (e.g., yogurt, hot dogs,"
            " supplement, beef sticks)."
        ),
    )
    flavor: str = Field(
        ...,
        description=(
            "Flavor if applicable. Empty if not found."
        ),
    )
    # Weight / Serving
    net_weight_oz: float = Field(
        ..., description="Net weight in ounces."
    )
    net_weight_g: float = Field(
        ..., description="Net weight in grams."
    )
    servings_per_container: int = Field(
        ..., description="Total servings per container."
    )
    serving_size: str = Field(
        ...,
        description="Serving size as printed (e.g., '1 stick (45g)').",
    )
    # Certifications (bool flags)
    is_organic: bool = Field(
        ..., description="True if labeled Organic / USDA Organic."
    )
    is_non_gmo: bool = Field(
        ..., description="True if Non-GMO Project Verified."
    )
    is_grass_fed: bool = Field(
        ..., description="True if labeled Grass-Fed."
    )
    is_kosher: bool = Field(
        ..., description="True if Kosher certified."
    )
    is_gluten_free: bool = Field(
        ..., description="True if labeled Gluten-Free."
    )
    # Dietary claims
    is_keto_friendly: bool = Field(
        ..., description="True if labeled Keto."
    )
    is_dairy_free: bool = Field(
        ..., description="True if labeled Dairy-Free."
    )
    has_no_added_sugar: bool = Field(
        ...,
        description="True if No Added Sugar / Zero Sugar.",
    )

Note: The full food label schema has 27 fields including

is_pasture_raised, is_certified_humane, no_antibiotics,

no_hormones, is_regenerative, is_paleo_friendly,

is_whole30_approved, is_lactose_free, usda_inspected, etc.

Add fields as needed for your use case.

---

6. CME / Continuing Education Certificate

from typing import Optional
from pydantic import BaseModel, Field


class CMECertificateSchema(BaseModel):
    recipient_name: str = Field(
        ..., description="Full name of the certificate recipient."
    )
    issuing_organization: str = Field(
        ...,
        description="Organization that issued the certificate.",
    )
    activity_title: str = Field(
        ...,
        description="Title of the educational activity or course.",
    )
    completion_date: Optional[str] = Field(
        None, description="Completion date (YYYY-MM-DD)."
    )
    credits_earned: float = Field(
        ..., description="Number of credits earned (numeric)."
    )
    credit_type: Optional[str] = Field(
        None,
        description=(
            "Type of credit (e.g., CME, CE, CPE, CLE)."
        ),
    )
    certificate_id: Optional[str] = Field(
        None, description="Certificate or confirmation number."
    )
    accreditation_statement: Optional[str] = Field(
        None,
        description="Accreditation or approval statement text.",
    )

---

7. Document Classification Schema

Generic classifier using Literal enum. Adapt the type list to your document mix.

from typing import Literal
from pydantic import BaseModel, Field


class DocTypeClassification(BaseModel):
    type: Literal[
        "invoice",
        "bank_statement",
        "pay_stub",
        "utility_bill",
        "receipt",
        "contract",
    ] = Field(
        description="The type of the document.",
        title="Document Type",
    )

Classify-then-Extract Pattern

from landingai_ade import LandingAIADE
from landingai_ade.lib import pydantic_to_json_schema

schema_map = {
    "invoice": InvoiceSchema,
    "bank_statement": BankStatementSchema,
    "pay_stub": PayStubSchema,
    "utility_bill": UtilityBillSchema,
}

client = LandingAIADE()
parse_result = client.parse(document=path)

# Step 1: Classify
cls_result = client.extract(
    schema=pydantic_to_json_schema(DocTypeClassification),
    markdown=parse_result.markdown,
)
doc_type: str = cls_result.extraction["type"]

# Step 2: Extract with type-specific schema
schema_cls = schema_map[doc_type]
extract_result = client.extract(
    schema=pydantic_to_json_schema(schema_cls),
    markdown=parse_result.markdown,
)

---

Tips for Custom Schemas

1. Use descriptive `description` fields — ADE uses them as extraction instructions. Be specific about format expectations. 2. Mark optional fields with Optional[T] and default=None to avoid extraction failures when a field is absent. 3. Use `default_factory=list` for array fields (line items, transactions) to avoid null results. 4. Keep nesting to one level — ADE supports TopLevel.nested_object but not TopLevel.nested.deeper. 5. Use `Literal` for enums — constrains extraction to known values. 6. Prefer `float` over `str` for monetary amounts — easier downstream processing. Keep a _raw string variant if you need the original format.

Table Stitching — Multi-Page Table Extraction

When a table spans multiple pages, ADE emits separate chunks per page and may represent some pages as plain text instead of <table> HTML. This inconsistency can occur on any page — not just the last one. This reference covers three approaches to stitch those chunks into a single output, generalized for any document type.

---

Decision Guide

---

Approach A — Parse + Extract (LLM-based)

The simplest and most robust approach. Parse the document, then call client.extract() with a schema that describes the full table as a List[RowModel]. The LLM reads the entire markdown — including any pages where the table was emitted as plain text — and returns structured JSON.

Schema design for multi-page tables

from typing import List, Optional
from pydantic import BaseModel, Field


class TableRow(BaseModel):
    """Customize fields for your specific table."""
    key_column: str = Field(
        description=(
            "Primary identifier (e.g., date, ID, row number). "
            "Use empty string for continuation rows."
        )
    )
    description: str = Field(
        description="Description or label column."
    )
    amount_a: Optional[str] = Field(
        default=None,
        description=(
            "First amount column (digits and commas only, "
            "no currency symbol). "
            "Null if not applicable for this row."
        ),
    )
    amount_b: Optional[str] = Field(
        default=None,
        description=(
            "Second amount column. "
            "Null if not applicable."
        ),
    )
    running_total: str = Field(
        description="Running total or balance after this row."
    )


class DocumentWithTable(BaseModel):
    rows: List[TableRow] = Field(
        description=(
            "All data rows in order across ALL pages of the "
            "document. Include rows from every page of the "
            "table, even if some pages render as plain text "
            "rather than tables. "
            "Skip column-header rows and section-header rows."
        )
    )

Key schema tips

• Say "across ALL pages" in the List field description — this

tells the LLM to look beyond the first table chunk.

• Mention "even if some pages render as plain text" — the LLM

will then scan text chunks for table-like content.

• Say "Skip column-header rows" — continued tables often repeat

headers on each page.

• Use `Optional[str]` for mutually exclusive columns — e.g., when

only one of "debit" or "credit" applies per row.

• Use `str` (not `float`) for amounts to preserve original

formatting (commas, decimals). Convert downstream if needed.

Implementation pattern

import json
from pathlib import Path

from landingai_ade import LandingAIADE
from landingai_ade.lib import pydantic_to_json_schema

client = LandingAIADE()

# Step 1: Parse (cache the result to avoid re-parsing)
parse_json = Path("output/parsed.json")
if parse_json.exists():
    data = json.loads(parse_json.read_text())
    markdown = data["markdown"]
else:
    pr = client.parse(document=Path("document.pdf"))
    parse_json.parent.mkdir(parents=True, exist_ok=True)
    parse_json.write_text(
        json.dumps(pr.model_dump(), indent=2, default=str)
    )
    markdown = pr.markdown

# Step 2: Extract with the multi-page table schema
er = client.extract(
    schema=pydantic_to_json_schema(DocumentWithTable),
    markdown=markdown,
)
rows = er.extraction["rows"]

Pros / Cons

• ✅ Handles all pages uniformly, including plain-text edge cases
• ✅ No custom parsing code
• ❌ Two ADE API calls → ~2× credit cost
• ❌ Slower: two network round-trips

---

Approach B — HTML Table Parsing (parse-only, fragile)

Reuse the cached parse result (zero extra credits). Parse <table> elements from the ADE markdown, detect which tables belong to the target table, merge their rows, then fall back to regex for any plain-text rows.

Warning: This approach requires strong similarity between rows

to write reliable detection and extraction regex. It is brittle when

table format varies across documents or even across pages of the same

document. Always validate against Approach A on a sample before

relying on this in production.

Generic HTML table parser

from html.parser import HTMLParser


class TableParser(HTMLParser):
    """Extract all <table> elements as list-of-rows-of-cells."""

    def __init__(self) -> None:
        super().__init__()
        self._tables: list[list[list[str]]] = []
        self._tbl: list[list[str]] | None = None
        self._row: list[str] | None = None
        self._cell: list[str] | None = None

    def handle_starttag(self, tag: str, attrs: list) -> None:
        if tag == "table":
            self._tbl = []
        elif tag == "tr" and self._tbl is not None:
            self._row = []
        elif tag == "td" and self._row is not None:
            self._cell = []

    def handle_endtag(self, tag: str) -> None:
        if tag == "table" and self._tbl is not None:
            self._tables.append(self._tbl)
            self._tbl = None
        elif tag == "tr" and self._tbl is not None:
            if self._row is not None:
                self._tbl.append(self._row)
            self._row = None
        elif tag == "td" and self._row is not None:
            if self._cell is not None:
                self._row.append(
                    "".join(self._cell).strip()
                )
            self._cell = None

    def handle_data(self, data: str) -> None:
        if self._cell is not None:
            self._cell.append(data)

    @property
    def tables(self) -> list[list[list[str]]]:
        return self._tables


def extract_html_tables(
    markdown: str,
) -> list[list[list[str]]]:
    p = TableParser()
    p.feed(markdown)
    return p.tables

Table detection strategies

Column count alone is often insufficient — multiple table types may share the same number of columns. Use content-based signals:

Is there a header row with known column names?
  ├─ YES → Match on header content (most reliable)
  │        e.g., cells[0]=="Date" and "Description" in cells[1]
  └─ NO → Does the first data cell match a known pattern?
       ├─ YES → Match on first-column pattern
       │        e.g., regex for "Mon DD" date format
       └─ NO → Use column count + content heuristics
                (least reliable — last resort)

Row filtering

After detecting the target table, filter out non-data rows:

• Column-header rows — repeated on each page (e.g., cells[0] == "Date")
• Section sub-headers — account names, category labels
• Summary/totals rows — may need special handling

Plain-text fallback

When ADE emits table rows as a text chunk, use regex to extract them. This is the most fragile part — it requires the text to have a predictable structure:

import re

# Example: lines starting with a date pattern
DATE_RE = re.compile(
    r"^((?:Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec)"
    r"\s+\d{1,2})\s+(.*)"
)
AMOUNT_RE = re.compile(r"[\d,]+\.\d{2}")


def parse_text_rows(
    text: str,
) -> list[dict[str, str]]:
    """Extract rows from plain-text table content.

    Customize the date/amount patterns for your document type.
    """
    rows: list[dict[str, str]] = []
    for line in text.splitlines():
        m = DATE_RE.match(line.strip())
        if not m:
            continue
        date_str, rest = m.group(1), m.group(2)
        amounts = AMOUNT_RE.findall(rest)
        desc = AMOUNT_RE.sub("", rest).strip()
        rows.append({
            "date": date_str,
            "description": desc,
            "amounts": amounts,
        })
    return rows

Domain-specific semantic checks

After stitching, add validation that leverages domain knowledge to both confirm correctness and resolve ambiguity:

These checks are especially valuable when plain-text fallback produces amounts that could belong to multiple columns.

Pros / Cons

• ✅ Single ADE API call → half the credit cost
• ✅ Fast (parse result is cached; parsing runs locally)
• ❌ Requires strong row similarity for reliable regex
• ❌ Brittle — format changes across documents break detection
• ❌ Plain-text fallback needs domain-specific validation

---

Approach C — pandas read_html (parse-only, quick)

Let pandas do the heavy lifting. Extract <table> HTML strings with a regex, feed each to pd.read_html(), use pandas signals to detect target tables, then pd.concat to merge.

Limitation: This approach cannot recover rows from non-table

chunks. If ADE emits some pages as plain text, those rows are lost.

Implementation pattern

import re
from io import StringIO

import pandas as pd

TABLE_RE = re.compile(r"<table\b[^>]*>.*?</table>", re.DOTALL)


def stitch_tables_pandas(
    markdown: str,
    expected_cols: int,
    date_pattern: str = (
        r"^(?:Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec)"
        r"\s+\d+"
    ),
) -> pd.DataFrame:
    """Extract and merge multi-page tables from ADE markdown.

    Args:
        markdown: Full ADE markdown output.
        expected_cols: Expected number of columns in the target table.
        date_pattern: Regex to match date-like values in column 0.
    """
    table_htmls = TABLE_RE.findall(markdown)
    target_dfs: list[pd.DataFrame] = []

    for html in table_htmls:
        df = pd.read_html(StringIO(html), thousands=",")[0]
        if df.shape[1] != expected_cols:
            continue
        col0 = df.iloc[:, 0]
        has_date = bool(
            col0.astype(str).str.match(date_pattern).any()
        )
        has_numeric = (
            df.select_dtypes(include="number").shape[1] >= 1
        )
        if has_date or has_numeric:
            target_dfs.append(df)

    if not target_dfs:
        raise ValueError("No target tables found")
    return pd.concat(target_dfs, ignore_index=True)

Detection signals from pandas

Pitfall: pandas 3.x NaN trap

After pd.read_html, empty cells are float('nan'). In pandas 3.x, after .astype(str), StringDtype keeps them as the float('nan') object rather than the string "nan". This means:

# BROKEN in pandas 3.x:
col_str = df.iloc[:, 0].astype(str)
mask = col_str == "nan"  # Always False!

# CORRECT — check on the raw column before string conversion:
mask = pd.isna(df.iloc[:, 0])

Amount formatting

pd.read_html(thousands=",") strips commas from numbers during dtype inference. If you need comma-formatted output in CSV:

def fmt_amount(val: object) -> str:
    if pd.isna(val):
        return ""
    if isinstance(val, float):
        return f"{val:,.2f}"
    s = str(val).strip()
    try:
        return f"{float(s):,.2f}"
    except ValueError:
        return s

Pros / Cons

• ✅ Least code — pd.read_html + pd.concat do most of the work
• ✅ No custom HTML parser or regex
• ✅ pandas dtype signals give a clear detection path
• ❌ Cannot recover rows from non-table chunks
• ❌ pd.NA / float('nan') trap in pandas 3.x
• ❌ thousands="," strips commas — must re-format for output

---

Common Pitfalls

1. ADE may emit any page as plain text — not just the last page. Always check chunk types across all pages during pre-flight.

2. Column count alone is insufficient — when multiple table types share the same column count, use content-based detection (header matching, first-column patterns, dtype signals).

3. Approach B regex is brittle — test on multiple documents before relying on it. Format variations across documents (or even across pages of the same document) will break detection.

4. Always validate with domain-specific semantic checks — these catch errors that structural parsing misses and resolve ambiguities in column assignment.

5. Cache parse results — save pr.model_dump() to JSON after the first parse. Load it for development instead of calling client.parse() again. Only re-parse when the document changes.

---

Pre-Flight Checklist for Table Stitching

Before choosing an approach, run the diagnostic parse and check:

Visualization Patterns

Patterns for visualizing ADE parse and extraction results: chunk image cropping, bounding box overlays, and word-level grounding highlights.

---

1. Chunk Image Extraction

Crop individual chunks from document pages using bounding box coordinates. Useful for QA, debugging, and building visual search indexes.

from pathlib import Path
from typing import Any, List, Optional

from PIL import Image

try:
    import pymupdf
except ImportError:
    pymupdf = None  # type: ignore[assignment]


def save_chunk_images(
    parse_result: Any,
    document_path: Path,
    output_dir: Path,
    zoom: float = 2.0,
) -> Optional[Path]:
    """Crop and save each chunk as a PNG image.

    Creates: output_dir/<doc_stem>/page_<N>/<type>.<id>.png

    Args:
        parse_result: from client.parse()
        document_path: original document file
        output_dir: base directory for chunk images
        zoom: render scale factor (2.0 = 144 DPI)

    Returns:
        Path to created document directory, or None on error.
    """
    if pymupdf is None:
        print("Install pymupdf: pip install pymupdf")
        return None

    doc_dir = output_dir / document_path.stem
    chunks = parse_result.chunks or []

    def _save_page_chunks(
        img: Image.Image,
        page_chunks: List[Any],
        page_num: int,
    ) -> None:
        w, h = img.size
        page_dir = doc_dir / f"page_{page_num}"
        page_dir.mkdir(parents=True, exist_ok=True)
        for ch in page_chunks:
            if (
                not hasattr(ch, "grounding")
                or ch.grounding.page != page_num
            ):
                continue
            box = ch.grounding.box
            crop = img.crop((
                int(box.left * w),
                int(box.top * h),
                int(box.right * w),
                int(box.bottom * h),
            ))
            fname = f"{ch.type}.{ch.id}.png"
            crop.save(page_dir / fname)

    try:
        if document_path.suffix.lower() == ".pdf":
            pdf = pymupdf.open(document_path)
            mat = pymupdf.Matrix(zoom, zoom)
            for page_num in range(len(pdf)):
                pix = pdf[page_num].get_pixmap(matrix=mat)
                img = Image.frombytes(
                    "RGB", [pix.width, pix.height], pix.samples
                )
                _save_page_chunks(img, chunks, page_num)
            pdf.close()
        else:
            img = Image.open(document_path).convert("RGB")
            _save_page_chunks(img, chunks, 0)
        return doc_dir
    except Exception as exc:
        print(f"Failed to save chunk images: {exc}")
        return None

Usage

from landingai_ade import LandingAIADE
from pathlib import Path

client = LandingAIADE()
pr = client.parse(document=Path("report.pdf"))
save_chunk_images(pr, Path("report.pdf"), Path("chunk_images/"))
# Creates: chunk_images/report/page_0/text.abc123.png, etc.

---

2. Grounding Overlay — Bounding Boxes on Pages

Draw color-coded bounding boxes on rendered page images to show where each chunk was detected.

from pathlib import Path
from typing import Any, Dict, Tuple

from PIL import Image, ImageDraw

try:
    import pymupdf
except ImportError:
    pymupdf = None  # type: ignore[assignment]

# Color map for chunk types (RGB tuples)
CHUNK_COLORS: Dict[str, Tuple[int, int, int]] = {
    "text": (40, 167, 69),         # green
    "table": (0, 123, 255),        # blue
    "marginalia": (111, 66, 193),  # purple
    "figure": (255, 0, 255),       # magenta
    "logo": (144, 238, 144),       # light green
    "card": (255, 165, 0),         # orange
    "attestation": (0, 255, 255),  # cyan
    "scan_code": (255, 193, 7),    # yellow
}
DEFAULT_COLOR = (200, 200, 200)


def render_page_image(
    document_path: Path,
    page_num: int,
    zoom: float = 2.0,
) -> Image.Image:
    """Render a single page as a PIL Image."""
    if document_path.suffix.lower() == ".pdf":
        if pymupdf is None:
            raise ImportError("pip install pymupdf")
        pdf = pymupdf.open(document_path)
        pix = pdf[page_num].get_pixmap(
            matrix=pymupdf.Matrix(zoom, zoom)
        )
        img = Image.frombytes(
            "RGB", [pix.width, pix.height], pix.samples
        )
        pdf.close()
        return img
    return Image.open(document_path).convert("RGB")


def annotate_page(
    img: Image.Image,
    chunks: list,
    page_num: int,
    line_width: int = 3,
) -> Image.Image:
    """Draw bounding boxes for all chunks on a page image."""
    annotated = img.copy()
    draw = ImageDraw.Draw(annotated)
    w, h = img.size

    for ch in chunks:
        if (
            not hasattr(ch, "grounding")
            or ch.grounding.page != page_num
        ):
            continue
        box = ch.grounding.box
        color = CHUNK_COLORS.get(
            getattr(ch, "type", ""), DEFAULT_COLOR
        )
        coords = [
            int(box.left * w),
            int(box.top * h),
            int(box.right * w),
            int(box.bottom * h),
        ]
        draw.rectangle(coords, outline=color, width=line_width)
    return annotated


def visualize_parse(
    parse_result: Any,
    document_path: Path,
    output_dir: Path,
    zoom: float = 2.0,
) -> None:
    """Render all pages with chunk bounding box overlays.

    Saves: output_dir/<doc_stem>/page_<N>_annotated.png
    """
    doc_dir = output_dir / document_path.stem
    doc_dir.mkdir(parents=True, exist_ok=True)
    chunks = parse_result.chunks or []

    # Determine page count
    if document_path.suffix.lower() == ".pdf":
        pdf = pymupdf.open(document_path)
        n_pages = len(pdf)
        pdf.close()
    else:
        n_pages = 1

    for page_num in range(n_pages):
        img = render_page_image(document_path, page_num, zoom)
        annotated = annotate_page(img, chunks, page_num)
        out_path = doc_dir / f"page_{page_num + 1}_annotated.png"
        annotated.save(out_path)

Visualize Extracted Fields Only

Show only the chunks that contributed to extracted fields (using extraction metadata references):

def visualize_extraction(
    parse_result: Any,
    extract_result: Any,
    document_path: Path,
    output_dir: Path,
    zoom: float = 2.0,
) -> None:
    """Draw boxes only for chunks referenced by extracted
    fields."""
    doc_dir = output_dir / document_path.stem
    doc_dir.mkdir(parents=True, exist_ok=True)

    # Collect referenced chunk IDs
    meta = getattr(extract_result, "extraction_metadata", {})
    if isinstance(meta, dict):
        ref_ids = set()
        for field_meta in meta.values():
            refs = (
                field_meta.get("references", [])
                if isinstance(field_meta, dict)
                else getattr(field_meta, "references", [])
            )
            ref_ids.update(refs)
    else:
        ref_ids = set()

    # Filter chunks to only referenced ones
    ref_chunks = [
        ch for ch in (parse_result.chunks or [])
        if getattr(ch, "id", None) in ref_ids
    ]

    if document_path.suffix.lower() == ".pdf":
        pdf = pymupdf.open(document_path)
        n_pages = len(pdf)
        pdf.close()
    else:
        n_pages = 1

    for page_num in range(n_pages):
        img = render_page_image(document_path, page_num, zoom)
        annotated = annotate_page(img, ref_chunks, page_num)
        out = doc_dir / f"page_{page_num + 1}_annotated.png"
        annotated.save(out)

---

3. Word-Level Grounding

Two approaches depending on whether the PDF contains native text or is scanned.

---

3a. Native PDF Word Search (preferred)

For text-based PDFs, PyMuPDF's get_text("words", clip=rect) finds words exactly with no OCR required. The key pattern is spatially restricting the search to specific ADE chunk bounding boxes, so occurrences in adjacent sections on the same page (e.g. an abstract above the introduction) are automatically excluded.

from pathlib import Path
from typing import Any, List, Tuple

from PIL import Image, ImageDraw

try:
    import pymupdf
except ImportError:
    pymupdf = None  # type: ignore[assignment]


def find_term_in_chunks(
    pdf_path: Path,
    page_num: int,
    chunks: list,           # ADE chunk objects with grounding
    term: str,
    zoom: float = 2.0,
) -> List[dict]:
    """Find *term* only within the ADE chunk bounding boxes on *page_num*.

    Uses PyMuPDF native text extraction clipped to each chunk rect so that
    occurrences outside the supplied chunks are ignored.

    Returns list of dicts: text, left, top, width, height (pixel coords
    at *zoom* scale, matching the image from render_page_image()).
    """
    if pymupdf is None:
        raise ImportError("pip install pymupdf")
    pdf = pymupdf.open(pdf_path)
    page = pdf[page_num]
    pw, ph = page.rect.width, page.rect.height

    boxes = []
    for ch in chunks:
        b = ch.grounding.box
        clip = pymupdf.Rect(b.left * pw, b.top * ph, b.right * pw, b.bottom * ph)
        # Each word entry: (x0, y0, x1, y1, "word", block_no, line_no, word_no)
        for x0, y0, x1, y1, text, *_ in page.get_text("words", clip=clip):
            if text.strip(".,;:!?()[]{}\"'–—") == term:
                boxes.append({
                    "text":   text,
                    "left":   int(x0 * zoom),
                    "top":    int(y0 * zoom),
                    "width":  int((x1 - x0) * zoom),
                    "height": int((y1 - y0) * zoom),
                })
    pdf.close()
    return boxes

Annotation and Redaction

The same annotate_page function handles both use cases — the only difference is the alpha value of the fill colour:

# Highlight: semi-transparent colour (text remains readable)
HIGHLIGHT = (255, 255, 0, 120)   # yellow

# Redact: opaque box (text is visually hidden)
REDACT = (0, 0, 0, 255)          # black, fully opaque


def annotate_page(
    page_img: Image.Image,
    boxes: List[dict],
    fill: Tuple[int, int, int, int] = HIGHLIGHT,
) -> Image.Image:
    """Overlay filled rectangles on a page image.

    Pass HIGHLIGHT for annotation or REDACT to cover sensitive content.
    Output is a PNG image — not a PDF-native redaction.
    """
    rgba = page_img.convert("RGBA")
    overlay = Image.new("RGBA", rgba.size, (0, 0, 0, 0))
    draw = ImageDraw.Draw(overlay)
    for b in boxes:
        draw.rectangle(
            [b["left"], b["top"],
             b["left"] + b["width"], b["top"] + b["height"]],
            fill=fill,
        )
    return Image.alpha_composite(rgba, overlay)

PDF-native redaction (permanently removes underlying text, not just

visually covers it) uses page.add_redact_annot() /

page.apply_redactions() from PyMuPDF. That does not depend on ADE and

belongs in the pdf skill.

Usage

from landingai_ade import LandingAIADE

client = LandingAIADE()
pr = client.parse(document=Path("paper.pdf"))

# Select only the chunks you want to search within
target_chunks = [ch for ch in pr.chunks if ch.grounding.page == 1
                 and "introduction" in (ch.markdown or "").lower()]

boxes = find_term_in_chunks(Path("paper.pdf"), page_num=1,
                            chunks=target_chunks, term="L2S")
img = render_page_image(Path("paper.pdf"), page_num=1)
highlighted = annotate_page(img, boxes, fill=HIGHLIGHT)
highlighted.convert("RGB").save("page_2_highlighted.png")

---

3b. OCR Word-Level Grounding (scanned PDFs)

For precise highlighting of extracted values within chunks. Uses Tesseract OCR on chunk crops + fuzzy matching to locate exact words.

Requires: pytesseract, tesseract system binary, fuzzywuzzy

from typing import List, Tuple

from PIL import Image, ImageDraw

try:
    import pytesseract
    from fuzzywuzzy import fuzz
    WORD_GROUNDING_AVAILABLE = True
except ImportError:
    WORD_GROUNDING_AVAILABLE = False


def find_words_in_chunk(
    chunk_image: Image.Image,
    search_text: str,
    confidence_threshold: int = 60,
    fuzzy_threshold: int = 80,
) -> List[dict]:
    """Find word-level bounding boxes matching search_text.

    Returns list of dicts with keys: text, left, top, width,
    height, conf, match_score.
    """
    if not WORD_GROUNDING_AVAILABLE:
        raise ImportError(
            "pip install pytesseract fuzzywuzzy python-Levenshtein"
        )

    ocr_data = pytesseract.image_to_data(
        chunk_image, output_type=pytesseract.Output.DICT
    )
    search_words = search_text.lower().split()
    matches: List[dict] = []

    for i, word in enumerate(ocr_data["text"]):
        conf = int(ocr_data["conf"][i])
        if conf < confidence_threshold or not word.strip():
            continue
        for sw in search_words:
            score = fuzz.ratio(word.lower(), sw)
            if score >= fuzzy_threshold:
                matches.append({
                    "text": word,
                    "left": ocr_data["left"][i],
                    "top": ocr_data["top"][i],
                    "width": ocr_data["width"][i],
                    "height": ocr_data["height"][i],
                    "conf": conf,
                    "match_score": score,
                })
    return matches


def highlight_words(
    chunk_image: Image.Image,
    matches: List[dict],
    color: Tuple[int, int, int, int] = (255, 255, 0, 100),
) -> Image.Image:
    """Draw semi-transparent highlights over matched words."""
    highlighted = chunk_image.convert("RGBA")
    overlay = Image.new("RGBA", highlighted.size, (0, 0, 0, 0))
    draw = ImageDraw.Draw(overlay)

    for m in matches:
        draw.rectangle(
            [
                m["left"],
                m["top"],
                m["left"] + m["width"],
                m["top"] + m["height"],
            ],
            fill=color,
        )
    return Image.alpha_composite(highlighted, overlay)

Full Word-Level Grounding Pipeline

def word_level_grounding(
    parse_result: Any,
    extract_result: Any,
    document_path: Path,
    output_dir: Path,
    zoom: float = 2.0,
) -> None:
    """For each extracted field, find and highlight the exact
    words in the source chunk.

    Saves highlighted chunk crops to output_dir.
    """
    output_dir.mkdir(parents=True, exist_ok=True)
    meta = getattr(extract_result, "extraction_metadata", {})
    extraction = extract_result.extraction

    # Build chunk lookup
    chunk_map = {
        ch.id: ch for ch in (parse_result.chunks or [])
        if hasattr(ch, "id")
    }

    for field_name, field_meta in (
        meta.items() if isinstance(meta, dict) else []
    ):
        refs = (
            field_meta.get("references", [])
            if isinstance(field_meta, dict)
            else getattr(field_meta, "references", [])
        )
        if not refs:
            continue

        # Get the extracted value
        value = _dig_value(extraction, field_name)
        if not value or not isinstance(value, str):
            continue

        chunk_id = refs[0]
        chunk = chunk_map.get(chunk_id)
        if not chunk or not hasattr(chunk, "grounding"):
            continue

        # Crop the chunk from the page
        page_num = chunk.grounding.page
        page_img = render_page_image(
            document_path, page_num, zoom
        )
        box = chunk.grounding.box
        w, h = page_img.size
        crop = page_img.crop((
            int(box.left * w),
            int(box.top * h),
            int(box.right * w),
            int(box.bottom * h),
        ))

        # Find and highlight words
        matches = find_words_in_chunk(crop, str(value))
        if matches:
            highlighted = highlight_words(crop, matches)
            out = output_dir / f"{field_name}_{chunk_id}.png"
            highlighted.save(out)


def _dig_value(d: dict, dotted_key: str) -> Any:
    """Get value from nested dict using __ separator."""
    parts = dotted_key.split("__")
    obj: Any = d
    for p in parts:
        if isinstance(obj, dict):
            obj = obj.get(p)
        else:
            return None
    return obj

---

Dependencies

# Chunk images + bounding box overlays + native word search (Sections 1–3a)
pip install landingai-ade Pillow pymupdf

# OCR word-level grounding / scanned PDFs (Section 3b)
pip install landingai-ade Pillow pymupdf pytesseract fuzzywuzzy python-Levenshtein
# Also requires: brew install tesseract (macOS) or apt install tesseract-ocr (Linux)