Building an LLM-Powered Data Extraction Pipeline: Schema Mapping, Validation Chains, and Handling Messy Real-World Documents

The extraction problem looks simple from the outside: given a PDF invoice, return a structured object with the vendor name, line items, totals, and payment terms. A quick demo with a clean, machine-generated PDF and function calling works in an hour.

Then you see what the production data actually looks like. Scanned invoices from the 1990s. Contracts where the party names appear in four different formats throughout the document. Medical records where a field labeled “Date of Service” might mean the admission date, the discharge date, or a billing date depending on which hospital system generated it. PDFs that are actually images. PDFs where the text layer is present but misaligned with the visual layout by three columns.

This article is about building an extraction pipeline that handles that reality: document preprocessing, schema-driven extraction with function calling, multi-layer validation chains, failure recovery, and cost management. The architecture scales from hundreds of documents per day to millions.

Why Regex and Traditional NER Fall Short

Before covering the LLM approach, it is worth being precise about why simpler alternatives break down, because the tradeoffs determine when to use each.

Regex extraction works reliably when the document format is fully controlled. US Social Security numbers, credit card numbers, standardized date formats, ISBN codes. It breaks on any variation the regex did not anticipate. A regex for invoice totals written for one vendor’s format will miss the same field on every other vendor’s invoices. Maintaining a regex library across hundreds of vendors is an engineering problem that grows without bound.

Traditional NER (named entity recognition with models like spaCy or fine-tuned BERT) is stronger than regex for entities like person names, organizations, locations, and dates, but it operates at the token level and has no concept of document structure or field relationships. It will find every date in a contract but cannot tell you which date is the effective date versus the expiration date versus the signing date. Training custom NER models for each document type is feasible but expensive in labeled data and maintenance overhead.

LLMs understand document structure, context, and field relationships at a level that neither regex nor NER approaches. They generalize across vendors and formats without retraining. The tradeoffs are cost, latency, and the need for a robust validation layer because LLMs can hallucinate values or apply wrong field mappings.

The practical answer for most production systems: use regex for high-confidence, format-stable fields (document IDs, dates in known formats, total amounts with known currency markers), and LLM extraction for everything that requires contextual understanding. The cost profile is better and the failure modes are more predictable.

Approach	Accuracy	Generalization	Maintenance	Cost per doc	Best for
Regex	High (when matched)	None	High (per format)	Minimal	Format-stable fields across known templates
Traditional NER	Medium	Low	Medium (retraining)	Low	General entity types (names, orgs, dates)
LLM extraction	High	Strong	Low	Medium to high	Complex relationships, varied formats, novel documents
Hybrid (regex + LLM)	Highest	Strong	Medium	Low to medium	Production at scale

Document Preprocessing

The extraction pipeline begins before the LLM sees anything. Document quality determines extraction quality, and production documents are rarely clean.

import { PDFDocument } from "pdf-lib";
import Tesseract from "tesseract.js";
import sharp from "sharp";

interface PreprocessedDocument {
  text: string;
  pageCount: number;
  isScanned: boolean;
  ocrConfidence?: number;
  chunks: DocumentChunk[];
}

interface DocumentChunk {
  index: number;
  text: string;
  pageStart: number;
  pageEnd: number;
  tokenEstimate: number;
}

async function preprocessDocument(
  buffer: Buffer,
  mimeType: string
): Promise<PreprocessedDocument> {
  if (mimeType === "application/pdf") {
    return preprocessPdf(buffer);
  }
  if (mimeType.startsWith("image/")) {
    return preprocessImage(buffer);
  }
  throw new Error(`Unsupported document type: ${mimeType}`);
}

async function preprocessPdf(buffer: Buffer): Promise<PreprocessedDocument> {
  const pdfDoc = await PDFDocument.load(buffer);
  const pageCount = pdfDoc.getPageCount();

  // Attempt text extraction first -- many PDFs have an embedded text layer
  const textByPage = await extractTextFromPdf(buffer);
  const fullText = textByPage.join("\n\n");

  // Heuristic: if we extracted fewer than 50 characters per page on average,
  // the PDF is likely scanned and the text layer is unreliable
  const avgCharsPerPage = fullText.length / pageCount;
  const isScanned = avgCharsPerPage < 50;

  if (isScanned) {
    return preprocessScannedPdf(buffer, pageCount);
  }

  return {
    text: fullText,
    pageCount,
    isScanned: false,
    chunks: chunkDocument(fullText, pageCount),
  };
}

async function preprocessScannedPdf(
  buffer: Buffer,
  pageCount: number
): Promise<PreprocessedDocument> {
  // Render each page to an image, then OCR
  const pageTexts: string[] = [];
  let totalConfidence = 0;

  for (let i = 0; i < pageCount; i++) {
    const pageImage = await renderPdfPageToImage(buffer, i);

    // Pre-process image for better OCR: deskew, increase contrast, remove noise
    const enhancedImage = await sharp(pageImage)
      .greyscale()
      .normalize()
      .sharpen()
      .toBuffer();

    const { data } = await Tesseract.recognize(enhancedImage, "eng", {
      logger: () => {}, // suppress progress logs in production
    });

    pageTexts.push(data.text);
    totalConfidence += data.confidence;
  }

  const fullText = pageTexts.join("\n\n");

  return {
    text: fullText,
    pageCount,
    isScanned: true,
    ocrConfidence: totalConfidence / pageCount,
    chunks: chunkDocument(fullText, pageCount),
  };
}

function chunkDocument(
  text: string,
  pageCount: number
): DocumentChunk[] {
  // Target: ~3000 tokens per chunk with overlap, respecting page boundaries
  const TARGET_CHUNK_TOKENS = 3000;
  const OVERLAP_TOKENS = 300;
  const CHARS_PER_TOKEN_ESTIMATE = 4;

  const targetChunkChars = TARGET_CHUNK_TOKENS * CHARS_PER_TOKEN_ESTIMATE;
  const overlapChars = OVERLAP_TOKENS * CHARS_PER_TOKEN_ESTIMATE;

  if (text.length <= targetChunkChars) {
    return [
      {
        index: 0,
        text,
        pageStart: 1,
        pageEnd: pageCount,
        tokenEstimate: Math.ceil(text.length / CHARS_PER_TOKEN_ESTIMATE),
      },
    ];
  }

  const chunks: DocumentChunk[] = [];
  let offset = 0;
  let chunkIndex = 0;

  while (offset < text.length) {
    const end = Math.min(offset + targetChunkChars, text.length);
    // Find a paragraph boundary near the end to avoid splitting mid-sentence
    const boundarySearch = text.lastIndexOf("\n\n", end);
    const chunkEnd =
      boundarySearch > offset + targetChunkChars * 0.7
        ? boundarySearch
        : end;

    const chunkText = text.slice(offset, chunkEnd);
    chunks.push({
      index: chunkIndex++,
      text: chunkText,
      pageStart: Math.floor((offset / text.length) * pageCount) + 1,
      pageEnd: Math.floor((chunkEnd / text.length) * pageCount) + 1,
      tokenEstimate: Math.ceil(chunkText.length / CHARS_PER_TOKEN_ESTIMATE),
    });

    offset = chunkEnd - overlapChars;
  }

  return chunks;
}

The OCR confidence score is worth tracking. If a document comes in at 40% OCR confidence, extraction results should be treated as low-confidence regardless of what the LLM returns. That signal belongs in your output metadata.

Schema-Driven Extraction with Function Calling

The extraction layer uses LLM function calling to map document content onto your target schema. Define the schema once with Zod, derive types from it, and hand the JSON Schema representation to the model.

import OpenAI from "openai";
import { z } from "zod";

const client = new OpenAI();

// Invoice extraction schema -- realistic complexity
const LineItemSchema = z.object({
  description: z.string(),
  quantity: z.number().positive(),
  unit_price: z.number().nonnegative(),
  total: z.number().nonnegative(),
  product_code: z.string().optional(),
});

const InvoiceSchema = z.object({
  invoice_number: z.string(),
  invoice_date: z.string().describe("ISO 8601 date string"),
  due_date: z.string().optional().describe("ISO 8601 date string if present"),
  vendor_name: z.string(),
  vendor_address: z.string().optional(),
  buyer_name: z.string(),
  buyer_address: z.string().optional(),
  line_items: z.array(LineItemSchema),
  subtotal: z.number().nonnegative(),
  tax_amount: z.number().nonnegative().optional(),
  total_amount: z.number().positive(),
  currency: z.string().length(3).describe("ISO 4217 currency code, e.g. USD"),
  payment_terms: z.string().optional(),
  // Extraction metadata -- not from the document, set by the pipeline
  extraction_confidence: z.number().min(0).max(1),
  fields_inferred: z.array(z.string()),
});

type Invoice = z.infer<typeof InvoiceSchema>;

interface ExtractionResult {
  success: boolean;
  data?: Invoice;
  rawOutput?: unknown;
  error?: string;
  method: "function_calling" | "json_mode" | "fallback" | "failed";
}

async function extractInvoice(
  documentText: string,
  documentId: string
): Promise<ExtractionResult> {
  const systemPrompt = `You are a document extraction system. Extract invoice data exactly as it appears in the document.
- If a field is not present, omit it rather than guessing.
- For dates, convert to ISO 8601 format (YYYY-MM-DD).
- For currency, use the ISO 4217 code. If ambiguous, use USD.
- Set extraction_confidence between 0 and 1 based on how clearly the document presents the data.
- List any fields in fields_inferred that you inferred from context rather than read directly.
- Do not hallucinate values. If the invoice number is not visible, omit it.`;

  const response = await client.chat.completions.create({
    model: "gpt-4o",
    tools: [
      {
        type: "function",
        function: {
          name: "extract_invoice",
          description: "Extract structured invoice data from document text",
          parameters: buildJsonSchema(InvoiceSchema),
        },
      },
    ],
    tool_choice: {
      type: "function",
      function: { name: "extract_invoice" },
    },
    messages: [
      { role: "system", content: systemPrompt },
      {
        role: "user",
        content: `Extract invoice data from this document:\n\n${documentText}`,
      },
    ],
  });

  const toolCall = response.choices[0].message.tool_calls?.[0];
  if (!toolCall) {
    return {
      success: false,
      error: "Model did not return a tool call",
      method: "function_calling",
    };
  }

  const rawArgs = JSON.parse(toolCall.function.arguments);
  return { success: true, rawOutput: rawArgs, method: "function_calling" };
}

The fields_inferred array is a deliberate design choice. Asking the model to self-report which fields it inferred versus read directly gives you a signal for downstream validation: inferred fields need higher scrutiny. An invoice number that was inferred is far more likely to be wrong than one the model found verbatim in the text.

Multi-Layer Validation Chains

Extraction and validation are separate stages. The LLM produces raw output; your pipeline validates it in layers before treating it as reliable.

interface ValidationResult {
  valid: boolean;
  errors: ValidationError[];
  warnings: ValidationWarning[];
  confidence: number; // adjusted after validation
}

interface ValidationError {
  field: string;
  code: string;
  message: string;
  severity: "blocking" | "recoverable";
}

interface ValidationWarning {
  field: string;
  code: string;
  message: string;
}

async function validateInvoice(
  raw: unknown,
  ocrConfidence?: number
): Promise<{ invoice?: Invoice; validation: ValidationResult }> {
  const errors: ValidationError[] = [];
  const warnings: ValidationWarning[] = [];

  // Layer 1: Schema validation -- types and required fields
  const parseResult = InvoiceSchema.safeParse(raw);
  if (!parseResult.success) {
    return {
      validation: {
        valid: false,
        errors: parseResult.error.issues.map((issue) => ({
          field: issue.path.join("."),
          code: "SCHEMA_VIOLATION",
          message: issue.message,
          severity: "blocking",
        })),
        warnings,
        confidence: 0,
      },
    };
  }

  const invoice = parseResult.data;

  // Layer 2: Business rule validation -- cross-field consistency
  const lineItemsTotal = invoice.line_items.reduce(
    (sum, item) => sum + item.total,
    0
  );

  // Line item totals should sum to subtotal within 1% tolerance
  const subtotalDiff = Math.abs(lineItemsTotal - invoice.subtotal);
  const subtotalTolerance = invoice.subtotal * 0.01;
  if (subtotalDiff > subtotalTolerance && subtotalDiff > 0.01) {
    errors.push({
      field: "subtotal",
      code: "TOTAL_MISMATCH",
      message: `Line item sum (${lineItemsTotal.toFixed(2)}) does not match subtotal (${invoice.subtotal.toFixed(2)})`,
      severity: "recoverable",
    });
  }

  // Total = subtotal + tax
  const expectedTotal = invoice.subtotal + (invoice.tax_amount ?? 0);
  const totalDiff = Math.abs(invoice.total_amount - expectedTotal);
  if (totalDiff > 0.02) {
    errors.push({
      field: "total_amount",
      code: "TOTAL_CALCULATION_ERROR",
      message: `total_amount (${invoice.total_amount}) does not equal subtotal + tax (${expectedTotal.toFixed(2)})`,
      severity: "recoverable",
    });
  }

  // Due date cannot precede invoice date
  if (invoice.due_date && invoice.invoice_date) {
    const invoiceDate = new Date(invoice.invoice_date);
    const dueDate = new Date(invoice.due_date);
    if (dueDate < invoiceDate) {
      errors.push({
        field: "due_date",
        code: "DATE_SEQUENCE_ERROR",
        message: "due_date is before invoice_date",
        severity: "blocking",
      });
    }
  }

  // Layer 3: Confidence scoring
  let confidence = invoice.extraction_confidence;

  // Penalize for each inferred field
  confidence -= invoice.fields_inferred.length * 0.05;

  // Penalize for recoverable errors (something is off, but we have data)
  const recoverableErrors = errors.filter((e) => e.severity === "recoverable");
  confidence -= recoverableErrors.length * 0.1;

  // Penalize if OCR confidence was low
  if (ocrConfidence !== undefined && ocrConfidence < 70) {
    confidence -= (70 - ocrConfidence) / 100;
    warnings.push({
      field: "document",
      code: "LOW_OCR_CONFIDENCE",
      message: `OCR confidence was ${ocrConfidence.toFixed(1)}%. Extraction results may be unreliable.`,
    });
  }

  confidence = Math.max(0, Math.min(1, confidence));

  const blockingErrors = errors.filter((e) => e.severity === "blocking");

  return {
    invoice: blockingErrors.length === 0 ? invoice : undefined,
    validation: {
      valid: blockingErrors.length === 0,
      errors,
      warnings,
      confidence,
    },
  };
}

The three validation layers serve distinct purposes. Schema validation confirms the LLM produced the right shape. Business rule validation catches semantic errors the schema cannot express: totals that do not add up, dates in impossible sequence, line item counts that do not match the stated summary. Confidence scoring gives downstream consumers a signal for how much to trust the output.

Handling Extraction Failures and Fallbacks

No extraction pipeline achieves 100% success on messy real-world documents. Design the failure path as explicitly as the success path.

interface PipelineOutput {
  documentId: string;
  status: "success" | "low_confidence" | "partial" | "failed";
  data?: Invoice;
  validation: ValidationResult;
  requiresHumanReview: boolean;
  reviewReason?: string;
  processingMs: number;
}

async function runExtractionPipeline(
  documentId: string,
  buffer: Buffer,
  mimeType: string
): Promise<PipelineOutput> {
  const startedAt = Date.now();

  // Step 1: Preprocess
  const doc = await preprocessDocument(buffer, mimeType);

  // Step 2: Attempt extraction on each chunk, merge results
  // For short documents, one chunk is the whole document
  const chunkResults = await Promise.all(
    doc.chunks.map((chunk) => extractInvoice(chunk.text, documentId))
  );

  // Take the first successful extraction result
  // For multi-chunk documents, you would merge fields across chunks
  const bestResult = chunkResults.find((r) => r.success && r.rawOutput);

  if (!bestResult?.rawOutput) {
    // Extraction failed entirely: try once with a simpler prompt
    const retryResult = await extractWithFallbackPrompt(
      doc.text,
      documentId
    );

    if (!retryResult.success || !retryResult.rawOutput) {
      return {
        documentId,
        status: "failed",
        validation: {
          valid: false,
          errors: [
            {
              field: "document",
              code: "EXTRACTION_FAILED",
              message: "All extraction attempts returned no data",
              severity: "blocking",
            },
          ],
          warnings: [],
          confidence: 0,
        },
        requiresHumanReview: true,
        reviewReason: "Extraction failed -- model returned no structured data",
        processingMs: Date.now() - startedAt,
      };
    }
  }

  const rawData = bestResult?.rawOutput ?? (await extractWithFallbackPrompt(doc.text, documentId)).rawOutput;

  // Step 3: Validate
  const { invoice, validation } = await validateInvoice(
    rawData,
    doc.ocrConfidence
  );

  // Step 4: Decide output status
  const processingMs = Date.now() - startedAt;

  if (!validation.valid) {
    return {
      documentId,
      status: "partial",
      validation,
      requiresHumanReview: true,
      reviewReason: `Validation errors: ${validation.errors
        .map((e) => e.code)
        .join(", ")}`,
      processingMs,
    };
  }

  if (validation.confidence < 0.6) {
    return {
      documentId,
      status: "low_confidence",
      data: invoice,
      validation,
      requiresHumanReview: true,
      reviewReason: `Confidence score ${validation.confidence.toFixed(2)} below threshold`,
      processingMs,
    };
  }

  return {
    documentId,
    status: "success",
    data: invoice,
    validation,
    requiresHumanReview: false,
    processingMs,
  };
}

async function extractWithFallbackPrompt(
  text: string,
  documentId: string
): Promise<ExtractionResult> {
  // Simpler prompt, fewer required fields, more permissive
  // Used when the primary extraction returns nothing
  const response = await client.chat.completions.create({
    model: "gpt-4o",
    response_format: { type: "json_object" },
    messages: [
      {
        role: "system",
        content:
          "Extract whatever invoice fields you can find. Return only what is clearly present. Use keys: invoice_number, invoice_date, vendor_name, total_amount, currency.",
      },
      {
        role: "user",
        content: `Document:\n\n${text.slice(0, 6000)}`, // truncate for fallback
      },
    ],
  });

  const raw = response.choices[0].message.content;
  if (!raw) return { success: false, error: "Empty response", method: "json_mode" };

  try {
    return {
      success: true,
      rawOutput: JSON.parse(raw),
      method: "json_mode",
    };
  } catch {
    return { success: false, error: "JSON parse failed", method: "failed" };
  }
}

The requiresHumanReview flag is the interface between your automated pipeline and your review queue. Documents that are low-confidence but structurally valid can be used downstream with the confidence score as a filter. Documents that fail validation should go into a review queue rather than being silently dropped or passed through with bad data.

Cost Management

LLM extraction at scale has a real cost structure that is worth modeling before you ship.

The primary lever is selective extraction. Most documents have a mix of easy-to-extract fields (amounts, dates, IDs) that regex handles reliably, and hard fields (party names in varied formats, contextual field mappings) that require the LLM. Running the full document through the LLM for fields you could have gotten with a regex is waste.

interface HybridExtractionPlan {
  regexFields: Record<string, string>; // field -> extracted value
  llmFields: string[]; // fields to hand to LLM
  reducedText: string; // relevant sections only, not full document
}

function planExtraction(text: string): HybridExtractionPlan {
  const regexFields: Record<string, string> = {};
  const llmFields: string[] = [];

  // Extract format-stable fields with regex first
  const invoiceNumberMatch = text.match(
    /(?:invoice\s*(?:#|number|no\.?)\s*:?\s*)([A-Z0-9\-]{4,20})/i
  );
  if (invoiceNumberMatch) {
    regexFields.invoice_number = invoiceNumberMatch[1];
  } else {
    llmFields.push("invoice_number");
  }

  const totalMatch = text.match(
    /(?:total\s+(?:due|amount|)?\s*:?\s*)\$?\s*([\d,]+\.\d{2})/i
  );
  if (totalMatch) {
    regexFields.total_amount = totalMatch[1].replace(",", "");
  } else {
    llmFields.push("total_amount");
  }

  // Always use LLM for relational fields
  llmFields.push("line_items", "vendor_name", "buyer_name", "payment_terms");

  // For LLM extraction, send only the most relevant sections
  // rather than the entire document
  const relevantSections = extractRelevantSections(text);

  return {
    regexFields,
    llmFields,
    reducedText: relevantSections,
  };
}

function extractRelevantSections(text: string): string {
  const lines = text.split("\n");
  const relevant: string[] = [];

  // Keep lines that contain extraction-relevant patterns
  const relevantPatterns = [
    /invoice|bill|receipt/i,
    /date|due|terms/i,
    /total|subtotal|tax|amount/i,
    /vendor|supplier|from|bill\s+to|ship\s+to/i,
    /item|description|qty|quantity|price|unit/i,
    /\$[\d,]+\.?\d*/,
    /\d{1,2}[\/\-]\d{1,2}[\/\-]\d{2,4}/,
  ];

  for (const line of lines) {
    if (relevantPatterns.some((pattern) => pattern.test(line))) {
      relevant.push(line);
    }
  }

  // Limit to first 200 relevant lines to cap token usage
  return relevant.slice(0, 200).join("\n");
}

For high-volume pipelines, the second major cost lever is caching. Identical or near-identical documents appear more often than you expect: the same vendor’s invoice format across hundreds of submissions, the same contract template with different parties. A semantic cache keyed on a normalized document fingerprint can achieve meaningful hit rates and eliminate LLM calls entirely for recognized formats.

Track cost per document type and per extraction method in your metrics. You will find that 20% of document types account for 80% of cost, usually because they are long or require multiple chunks. Those are the candidates for custom preprocessing or regex augmentation.

Production Considerations

Log the raw LLM output before validation, always. When an extraction fails validation in production, you need to see exactly what the model returned. Discard-on-parse-failure makes root cause analysis nearly impossible. Store raw outputs with a correlation ID and document ID.

Track validation failure rates by error code, not just overall pass rate. TOTAL_MISMATCH errors cluster by vendor: one vendor’s invoices consistently fail because they include a deposit credit that your schema does not model. DATE_SEQUENCE_ERROR errors might cluster by region because your date parser assumes MM/DD/YYYY but European documents use DD/MM/YYYY. Aggregate by error code to find these patterns.

Set a confidence threshold and stick to it. The temptation is to lower the confidence threshold when you want higher throughput. Resist it. A lower threshold means more data with more errors flowing downstream. The correct response to low confidence is a better pipeline (better preprocessing, better prompts, better schema), not a lower bar.

Test with adversarial document variety before launch. Your happy path test set probably has clean, modern PDFs. Your production documents will include: fax-to-PDF scans, multi-column layouts the text extractor linearizes incorrectly, handwritten annotations over printed text, and foreign language documents with mixed-language field labels. Build a test corpus from your actual incoming document types before declaring the pipeline production-ready.

Version your schemas explicitly. When you add a required field or change a type, documents that were extracted with the old schema fail new validation. Use a schema version identifier in your stored outputs so you can distinguish “this field is missing because the document did not have it” from “this field is missing because it was extracted before we added it.”

The LLM extraction pipeline is not a replacement for domain knowledge about your documents. The models that work best are the ones where you have invested in prompt engineering specific to your document types, built validation rules that encode your actual business constraints, and tuned the confidence thresholds against real failure data. That investment is not glamorous, but it is what separates a demo from a system that runs reliably on the documents your users actually send.