Building a Data Extraction Pipeline with LlamaIndex: Complete Guide

Modern applications increasingly rely on turning unstructured data — PDFs, web pages, emails, and reports — into structured, queryable information. LlamaIndex is a leading framework for connecting custom data sources to large language models (LLMs), and one of its most powerful capabilities is building robust data extraction pipelines. In this guide, you'll learn how to design, implement, and optimize an end-to-end extraction pipeline using LlamaIndex.

What Is a LlamaIndex Data Extraction Pipeline?

A data extraction pipeline in LlamaIndex is a sequence of components that ingest raw documents, parse them into nodes, and use LLMs to extract structured information according to a predefined schema. Unlike simple RAG (Retrieval-Augmented Generation) pipelines that retrieve text chunks, extraction pipelines produce structured outputs — JSON objects, Pydantic models, or database records — that can be directly consumed by downstream systems.

The core building blocks include:

Document Loaders — read data from sources like files, URLs, databases, or APIs.
Node Parsers — split documents into manageable chunks (nodes).
Extractors — LLM-powered modules that pull structured fields from each node.
Schemas — Pydantic models that define the expected output structure.
Sinks — destinations for extracted data, such as databases, files, or APIs.

Why It Matters

Unstructured data accounts for an estimated 80% of enterprise information. Manually extracting fields from contracts, invoices, research papers, or support tickets is slow, error-prone, and unscalable. LlamaIndex's extraction pipelines offer several advantages:

Automation — convert thousands of documents into structured records without manual review.
Consistency — Pydantic schemas enforce type safety and validation across all outputs.
Flexibility — swap LLM providers, parsers, or loaders without rewriting business logic.
Observability — built-in tracing and callbacks help debug extraction failures.
Cost control — chunking strategies and targeted extraction reduce token usage.

Prerequisites and Installation

Before building the pipeline, install the required packages. This guide uses Python 3.10+ and assumes basic familiarity with Pydantic.

pip install llama-index llama-index-core \
  llama-index-llms-openai \
  llama-index-readers-file \
  pydantic python-dotenv

Create a .env file with your API key:

OPENAI_API_KEY=sk-your-key-here

Step 1: Defining the Extraction Schema

The schema is the contract between your pipeline and downstream systems. Use Pydantic to define exactly what fields you want to extract, including descriptions that guide the LLM.

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


class InvoiceItem(BaseModel):
    description: str = Field(..., description="Description of the product or service")
    quantity: float = Field(..., description="Number of units")
    unit_price: float = Field(..., description="Price per unit in USD")
    total: float = Field(..., description="Line total in USD")


class Invoice(BaseModel):
    invoice_number: str = Field(..., description="Unique invoice identifier")
    vendor_name: str = Field(..., description="Name of the vendor or supplier")
    invoice_date: Optional[date] = Field(None, description="Date the invoice was issued")
    due_date: Optional[date] = Field(None, description="Payment due date")
    items: List[InvoiceItem] = Field(default_factory=list, description="Line items on the invoice")
    subtotal: Optional[float] = Field(None, description="Subtotal before tax")
    tax_amount: Optional[float] = Field(None, description="Total tax amount")
    total_amount: float = Field(..., description="Final total amount due")
    currency: str = Field(default="USD", description="ISO currency code")

Field descriptions are critical. They tell the LLM exactly what to look for and reduce hallucinations. Always mark optional fields with Optional and provide sensible defaults.

Step 2: Loading Documents

LlamaIndex provides dozens of readers. For this example, we'll load PDF invoices from a local directory, but the same pattern works for web pages, Notion, Google Drive, or databases.

from llama_index.core import SimpleDirectoryReader

def load_documents(directory: str = "./invoices"):
    reader = SimpleDirectoryReader(
        input_dir=directory,
        required_exts=[".pdf", ".txt", ".docx"],
        recursive=True,
    )
    documents = reader.load_data()
    print(f"Loaded {len(documents)} document(s)")
    return documents

Each Document object contains the text content plus metadata such as file path and creation time. You can enrich metadata during loading:

documents = reader.load_data()
for doc in documents:
    doc.metadata.update({
        "source_type": "invoice",
        "ingested_at": datetime.now().isoformat(),
    })

Step 3: Parsing Documents into Nodes

Nodes are the atomic units LlamaIndex processes. For extraction, you typically want larger chunks that preserve context — a single invoice should ideally fit in one node.

from llama_index.core.node_parser import SentenceSplitter

def parse_nodes(documents):
    parser = SentenceSplitter(
        chunk_size=1024,
        chunk_overlap=100,
        paragraph_separator="\n\n",
    )
    nodes = parser.get_nodes_from_documents(documents)
    print(f"Created {len(nodes)} node(s)")
    return nodes

For structured documents like invoices, consider using TokenTextSplitter or a custom parser that respects document boundaries. If each file is a single invoice, you can skip splitting entirely and treat each document as one node.

Step 4: Configuring the LLM

Set up the LLM that will perform extraction. OpenAI's GPT-4o models work well for structured extraction, but you can substitute any LlamaIndex-supported provider.

from llama_index.llms.openai import OpenAI
from llama_index.core import Settings

llm = OpenAI(
    model="gpt-4o",
    temperature=0.0,
    max_tokens=4096,
)

Settings.llm = llm

Setting temperature=0.0 is essential for extraction — you want deterministic, factual outputs rather than creative variation.

Step 5: Building the Extractor

LlamaIndex provides PydanticProgramExtractor and the lower-level StructuredLLMExtractor for pulling structured data. The cleanest approach uses llm.structured_predict, which directly returns a validated Pydantic instance.

from llama_index.core.prompts import PromptTemplate

EXTRACTION_PROMPT = PromptTemplate(
    """You are a data extraction assistant. Extract invoice information
from the following document text. Return only the structured data.

If a field is not present in the document, leave it as null or empty.
Do not guess or fabricate values. Be precise with numbers and dates.

Document text:
{text}
"""
)

def extract_from_node(node, schema=Invoice):
    response = llm.structured_predict(
        schema=schema,
        prompt=EXTRACTION_PROMPT,
        text=node.get_content(),
    )
    return response

The structured_predict method handles prompt formatting, function calling, and Pydantic validation in one call. If the LLM output fails validation, LlamaIndex raises a ValidationError you can catch and retry.

Step 6: Assembling the Full Pipeline

Now combine all components into a reusable pipeline class with error handling and retry logic.

import json
import logging
from pathlib import Path
from typing import List, Dict, Any

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class InvoiceExtractionPipeline:
    def __init__(self, input_dir: str, output_path: str):
        self.input_dir = input_dir
        self.output_path = Path(output_path)
        self.llm = OpenAI(model="gpt-4o", temperature=0.0)
        Settings.llm = self.llm

    def run(self) -> List[Dict[str, Any]]:
        documents = load_documents(self.input_dir)
        nodes = parse_nodes(documents)

        results = []
        for i, node in enumerate(nodes):
            logger.info(f"Processing node {i + 1}/{len(nodes)}")
            try:
                invoice = extract_from_node(node)
                record = invoice.model_dump(mode="json")
                record["_source_file"] = node.metadata.get("file_path", "unknown")
                results.append(record)
            except Exception as e:
                logger.error(f"Failed on node {i + 1}: {e}")
                results.append({
                    "_source_file": node.metadata.get("file_path", "unknown"),
                    "_error": str(e),
                })

        self._save_results(results)
        return results

    def _save_results(self, results: List[Dict[str, Any]]):
        self.output_path.parent.mkdir(parents=True, exist_ok=True)
        with open(self.output_path, "w") as f:
            json.dump(results, f, indent=2, default=str)
        logger.info(f"Saved {len(results)} records to {self.output_path}")


if __name__ == "__main__":
    pipeline = InvoiceExtractionPipeline(
        input_dir="./invoices",
        output_path="./output/invoices_extracted.json",
    )
    extracted = pipeline.run()
    print(f"Extracted {len(extracted)} invoice records")

Step 7: Storing Extracted Data in a Database

JSON files are fine for prototyping, but production pipelines should write to a database. Here's how to persist extracted records using SQLAlchemy.

from sqlalchemy import create_engine, Column, String, Float, JSON
from sqlalchemy.orm import sessionmaker, declarative_base

Base = declarative_base()


class InvoiceRecord(Base):
    __tablename__ = "invoices"
    id = Column(String, primary_key=True)
    vendor_name = Column(String)
    total_amount = Column(Float)
    raw_data = Column(JSON)


def save_to_db(records: List[Dict[str, Any]], db_url: str = "sqlite:///invoices.db"):
    engine = create_engine(db_url)
    Base.metadata.create_all(engine)
    Session = sessionmaker(bind=engine)

    with Session() as session:
        for record in records:
            if "_error" in record:
                continue
            db_record = InvoiceRecord(
                id=record.get("invoice_number", record["_source_file"]),
                vendor_name=record.get("vendor_name"),
                total_amount=record.get("total_amount"),
                raw_data=record,
            )
            session.merge(db_record)
        session.commit()
    print(f"Committed {len(records)} records to database")

Best Practices

To build production-grade extraction pipelines, follow these guidelines:

Use precise schemas — every field should have a descriptive Field(description=...). Vague schemas produce vague extractions.
Set temperature to zero — extraction is a deterministic task. Creative LLM behavior introduces errors.
Handle failures gracefully — wrap extraction calls in try/except blocks and log failures with source metadata for later review.
Validate with Pydantic — leverage Pydantic validators to enforce business rules, such as checking that total_amount equals subtotal + tax_amount.
Batch strategically — process documents in batches to respect rate limits, and use async APIs for higher throughput.
Cache results — store intermediate extraction results so you don't re-run expensive LLM calls when debugging downstream code.
Use observability tools — integrate LlamaIndex's callback_manager or tools like Langfuse and Arize Phoenix to trace token usage and extraction quality.
Choose the right chunk size — for extraction, larger chunks that preserve full document context usually outperform small, fragmented chunks.
Test with edge cases — include multi-page invoices, handwritten notes, and documents with missing fields in your test suite.
Consider cost — monitor token consumption. If most invoices are short, a smaller model like GPT-4o-mini may suffice and cut costs significantly.

Adding a Human-in-the-Loop Review Step

For high-stakes extraction, add a confidence scoring and review queue. You can estimate confidence by asking the LLM to self-assess, or by comparing extracted totals against computed sums of line items.

def compute_confidence(invoice: dict) -> float:
    if not invoice.get("items"):
        return 0.3
    computed_total = sum(item["total"] for item in invoice["items"])
    stated_total = invoice.get("total_amount", 0)
    if stated_total == 0:
        return 0.5
    diff = abs(computed_total - stated_total) / stated_total
    if diff < 0.01:
        return 0.95
    elif diff < 0.05:
        return 0.75
    else:
        return 0.4

# Flag low-confidence records for manual review
for record in results:
    if "_error" not in record:
        record["_confidence"] = compute_confidence(record)
        if record["_confidence"] < 0.75:
            record["_needs_review"] = True

Conclusion

Building a data extraction pipeline with LlamaIndex gives you a flexible, production-ready way to transform unstructured documents into structured, actionable data. By combining well-designed Pydantic schemas, appropriate document loaders, careful chunking, and robust error handling, you can create pipelines that scale from a handful of test files to thousands of production documents. Start with the simple pipeline outlined here, then layer in database persistence, confidence scoring, observability, and async processing as your needs grow. The key to success is iterative refinement — test against real documents, monitor extraction quality, and continuously tighten your schemas and prompts until the output meets your accuracy requirements.