Build a Document Extraction Pipeline

Extract structured data from invoices using GPT-4o-mini and a QA agent.

Introduction

30 Second Summary

Every business processes stacks of invoices, receipts, and contracts. Extracting data from these documents by hand is slow and error-prone. Modern AI can read these documents visually, but it still makes mistakes. The real challenge is building a system that catches and fixes its own errors.

In this project, you will build a document extraction pipeline that reads invoice images, outputs structured data, and then reviews its own work through a QA agent that automatically detects and corrects extraction errors.

What You'll Build

A Python-based extraction system that uses GPT-4o-mini's vision capability to pull structured data from invoice images, validates results with deterministic checks, and self-corrects errors through an agentic QA loop.

By the end of this project, you'll have:

  • A vision-based extraction pipeline that converts invoice images into validated, structured JSON using OpenAI Structured Outputs and Pydantic schemas.
  • A set of deterministic validation rules that catch math errors, missing fields, and format issues without relying on the LLM.
  • A QA agent that flags low-confidence results and attempts automatic self-correction through targeted re-prompting.
  • Secret Mission: A Streamlit demo UI for drag-and-drop invoice processing and a batch processor with aggregate accuracy reporting.

Are there any prerequisites?

You should have basic Python knowledge (variables, functions, classes) and an OpenAI API key. The project will guide you through installing all required packages.

Before We Start

Before We Start...

Set Up Your Project Environment

Before you can extract data from invoices, you need a project environment with the right tools installed.

In this step, you will create a Python project folder, install the packages your pipeline depends on, and verify that your OpenAI API key works.

In this step, get ready to:
  • Create a project folder, virtual environment, and install packages.
  • Set your OpenAI API key and verify connectivity with a test call.
  • Download sample invoice images for testing.
Create your project folder and install packages

Your extraction pipeline needs three packages: the OpenAI Python SDK for calling GPT-4o-mini, Pydantic for defining extraction schemas, and Streamlit for the optional demo UI later.

  • Open your terminal.

How to open your terminal

Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar. Type Terminal (macOS) or PowerShell (Windows) and press Enter.

  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir doc-extraction-pipeline
cd doc-extraction-pipeline
  • Create a Python virtual environment by running:
python3 -m venv venv

What is a virtual environment?

A virtual environment is an isolated space for your Python project's packages. It keeps this project's dependencies separate from other Python projects on your machine.

  • Activate the virtual environment:

macOS

source venv/bin/activate

Windows

venv\Scripts\activate

You should see (venv) appear at the start of your terminal prompt.

  • Install the three packages your project needs by running:
pip install openai pydantic streamlit

You should see output showing each package downloading and installing. The final line will say Successfully installed followed by a list of package names.

Seeing a permission error or install failure?

Make sure (venv) is visible in your terminal prompt. If it is not, you need to activate the virtual environment first.

If you see a network error, check your internet connection and try running the install command again.

Help me fix my pip install error

  • Create the empty project files and the invoices directory by running:
touch schemas.py extractor.py validator.py qa_agent.py main.py
mkdir invoices
  • Confirm the files and folder exist by running:
ls

You should see extractor.py, invoices, main.py, qa_agent.py, schemas.py, validator.py, and venv in the output.

Set your OpenAI API key and verify connectivity

The OpenAI SDK automatically looks for an environment variable called OPENAI_API_KEY. You need to set this in your terminal session so the SDK can authenticate your API calls.

  • Set your API key as an environment variable by running this command, replacing the placeholder with your actual key:

macOS

export OPENAI_API_KEY=your-api-key-here

Windows

$env:OPENAI_API_KEY="your-api-key-here"

Find your API key

Your OpenAI API key starts with sk- and is available from the API keys page in your OpenAI dashboard. If you do not have one yet, click Create new secret key.

  • Verify your API key works by running this test command:
python3 -c "from openai import OpenAI; client = OpenAI(); r = client.responses.create(model='gpt-4o-mini', input='Say hello in 3 words'); print(r.output_text)"

You should see a short three-word greeting printed in your terminal. This confirms the SDK can reach the OpenAI API with your key.

Seeing an authentication error?

Double-check that you replaced your-api-key-here with your actual key. Make sure there are no extra spaces around the key.

If you see RateLimitError, check that your OpenAI account has billing set up and available credits.

Help me troubleshoot my OpenAI API key

Download sample invoice images for testing

Your extraction pipeline needs real invoice images to process. You will download a couple of sample invoices into your invoices/ directory.

  • Search Google Images for "sample invoice" and find 2 invoice images that are clear and readable. Look for invoices showing typical fields like vendor name, line items, and totals.
  • Right-click each image and select Save Image As.
  • Save them into the invoices/ folder inside your doc-extraction-pipeline folder on your Desktop.
  • Name them something simple like sample1.png and sample2.png.

What makes a good test invoice?

Look for invoices with clearly printed text, a table of line items with quantities and prices, and visible totals. Avoid handwritten invoices or blurry screenshots.

The OpenAI vision API supports PNG, JPEG, and WEBP formats. Each image must be under 50 MB.

  • Confirm your images are saved correctly by running:
ls invoices/

You should see your invoice image files listed in the output.

  • Open Cursor on your project folder.

How to open Cursor

Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar. Type Cursor and press Enter.

Once Cursor opens, click File in the top menu bar.

Click Open Folder. Navigate to your Desktop and select the doc-extraction-pipeline folder.

You should see all your project files listed in the Explorer sidebar on the left.

Your environment is ready. Next up, you will define the Pydantic schema that tells GPT-4o-mini exactly what data to extract from your invoices.

Define Your Extraction Schema with Pydantic

Your project folder is set up and your OpenAI connection is verified. Now you need to tell the AI model exactly what to extract from an invoice.

You will define Pydantic models that describe the shape of your extraction output. When you pass these models to the OpenAI API, it guarantees the response matches your schema exactly. No parsing, no guessing, no malformed JSON.

In this step, get ready to:
  • Define Pydantic models for invoice line items and invoice data with confidence scores.
  • Add validation and result models to track extraction quality.
  • Test that your schema serializes to valid JSON Schema.
Create your invoice extraction models

Every field you want the AI to extract needs a corresponding field in your Pydantic model. The OpenAI API converts your model into a JSON Schema and forces the LLM to produce output that conforms to it.

  • Open schemas.py in Cursor.
  • Add the imports and your first model by pasting the following code:
from typing import Optional
from pydantic import BaseModel


class InvoiceLineItem(BaseModel):
    description: str
    quantity: Optional[float] = None
    unit_price: Optional[float] = None
    amount: Optional[float] = None
    confidence: Optional[float] = None

What does this code do?

InvoiceLineItem represents a single row on an invoice (e.g. "Web Development Services - 10 hours @ $150").

  • description is required (str with no default). Every line item must have a description.
  • The other fields are Optional[float] with a default of None. If the model cannot find a value, it returns null instead of hallucinating.
  • confidence is a float between 0.0 and 1.0 where the model self-reports how certain it is about the extracted amount.

Now add the main invoice data model below InvoiceLineItem. This is the top-level schema that captures the full invoice.

  • Add the following class below your InvoiceLineItem model:
class InvoiceData(BaseModel):
    vendor_name: Optional[str] = None
    vendor_name_confidence: Optional[float] = None
    vendor_address: Optional[str] = None
    invoice_number: Optional[str] = None
    invoice_number_confidence: Optional[float] = None
    invoice_date: Optional[str] = None
    invoice_date_confidence: Optional[float] = None
    due_date: Optional[str] = None
    customer_name: Optional[str] = None
    line_items: list[InvoiceLineItem] = []
    subtotal: Optional[float] = None
    subtotal_confidence: Optional[float] = None
    tax_amount: Optional[float] = None
    tax_amount_confidence: Optional[float] = None
    total_amount: Optional[float] = None
    total_amount_confidence: Optional[float] = None
    currency: Optional[str] = None

Why pair every field with a confidence score?

Notice the pattern: vendor_name is paired with vendor_name_confidence. This gives your QA agent a signal for each extracted value. Fields with low confidence (below 0.85) get flagged for re-extraction.

The model self-reports these scores, so they are not perfectly calibrated. Combined with the deterministic validation checks you will build later, they form a reliable quality signal for routing documents to human review.

  • Save schemas.py.

Seeing a red underline in Cursor?

  • Check that your imports at the top include both Optional from typing and BaseModel from pydantic.
  • Make sure your virtual environment is selected as the Python interpreter in Cursor. Click the Python version in the bottom status bar and select the one from your venv folder.

Why is Pydantic showing import errors in my editor?

Add validation and result models

Extracting data is only half the job. You also need models to track whether the extraction passed quality checks and what corrections were made.

  • Add the following three classes at the bottom of schemas.py, below InvoiceData:
class ValidationCheck(BaseModel):
    name: str
    status: str  # "pass", "fail", "warning"
    message: str


class ValidationResult(BaseModel):
    checks: list[ValidationCheck] = []
    is_valid: bool = True
    flagged_fields: list[str] = []


class ExtractionResult(BaseModel):
    invoice: InvoiceData
    validation: ValidationResult
    attempts: int = 1
    corrections_made: list[str] = []
    overall_confidence: float = 0.0

What do these models track?

  • ValidationCheck stores one pass/fail/warning result with a human-readable message (e.g. "Line items sum does not match subtotal").
  • ValidationResult collects all checks and lists which fields need attention in flagged_fields.
  • ExtractionResult wraps everything together: the extracted invoice, validation outcome, number of QA correction attempts, and an overall confidence score.
  • Save schemas.py.

Getting a type error on list[ValidationCheck]?

  • The list[ValidationCheck] syntax requires Python 3.9 or higher. If you are on an older version, you would need List[ValidationCheck] from typing. Check your version by running python --version in your terminal.

Help me fix type annotation errors in my Pydantic schemas.

Test your schema serialization

OpenAI Structured Outputs works by converting your Pydantic model into a JSON Schema. You can verify this conversion works correctly before sending anything to the API.

  • Test the schema serialization by running this command in your terminal:
python -c "from schemas import InvoiceData; import json; print(json.dumps(InvoiceData.model_json_schema(), indent=2))"

You should see a JSON Schema output with all your fields listed under properties, including nested definitions for InvoiceLineItem. This is the exact schema that the OpenAI API will use to constrain its output.

Seeing ModuleNotFoundError?

  • Make sure you are running the command from your project folder (the one containing schemas.py).
  • Confirm your virtual environment is activated. You should see (venv) at the start of your terminal prompt.

Why can't Python find my schemas module?

✔️ Awesome, I've got everything!

Great. Double check you have saved schemas.py and your terminal printed valid JSON Schema output.

ⓧ I'd like to double check the full code

Here is the complete schemas.py file. Make sure yours matches exactly.

from typing import Optional
from pydantic import BaseModel


class InvoiceLineItem(BaseModel):
    description: str
    quantity: Optional[float] = None
    unit_price: Optional[float] = None
    amount: Optional[float] = None
    confidence: Optional[float] = None


class InvoiceData(BaseModel):
    vendor_name: Optional[str] = None
    vendor_name_confidence: Optional[float] = None
    vendor_address: Optional[str] = None
    invoice_number: Optional[str] = None
    invoice_number_confidence: Optional[float] = None
    invoice_date: Optional[str] = None
    invoice_date_confidence: Optional[float] = None
    due_date: Optional[str] = None
    customer_name: Optional[str] = None
    line_items: list[InvoiceLineItem] = []
    subtotal: Optional[float] = None
    subtotal_confidence: Optional[float] = None
    tax_amount: Optional[float] = None
    tax_amount_confidence: Optional[float] = None
    total_amount: Optional[float] = None
    total_amount_confidence: Optional[float] = None
    currency: Optional[str] = None


class ValidationCheck(BaseModel):
    name: str
    status: str  # "pass", "fail", "warning"
    message: str


class ValidationResult(BaseModel):
    checks: list[ValidationCheck] = []
    is_valid: bool = True
    flagged_fields: list[str] = []


class ExtractionResult(BaseModel):
    invoice: InvoiceData
    validation: ValidationResult
    attempts: int = 1
    corrections_made: list[str] = []
    overall_confidence: float = 0.0

Your extraction schema is locked in. Next up, you will build the vision-based extractor that sends invoice images to GPT-4o-mini and gets structured data back using these exact models.

Build the Vision-Based Extractor

Your Pydantic schemas define exactly what invoice data looks like. Now it's time to use those schemas to extract real data from invoice images.

GPT-4o-mini can read images directly and return structured JSON that matches your schema. In this step, you'll build the extractor that sends an invoice image to the model and gets back typed, validated data using OpenAI Structured Outputs.

In this step, get ready to:
  • Define the system prompt for invoice extraction.
  • Build the extraction function using the OpenAI Responses API.
  • Test the extractor against a sample invoice image.
Add the system prompt and imports

The extraction prompt tells GPT-4o-mini exactly how to behave when it reads an invoice image. It instructs the model to extract visible text, mark unknown fields as null, and self-report confidence scores for each field.

  • Open extractor.py and paste the following imports and system prompt:
import base64
import sys
from pathlib import Path

from openai import OpenAI
from schemas import InvoiceData


EXTRACTION_PROMPT = """You are a document extraction specialist. Extract all invoice data from this image.

Rules:
- Extract only text values visible in the document.
- If a field is not found, set it to null.
- For each field, provide a confidence score between 0.0 and 1.0.
- For line items, extract description, quantity, unit_price, and amount.
- Amounts should be numeric values without currency symbols.
- Dates should be in the format found in the document (do not normalize).
"""

What does this code do?

The imports bring in base64 for image encoding, Path for file extension detection, and the OpenAI client for API calls.

The EXTRACTION_PROMPT is a multi-line string that serves as the model's instructions. Each rule eliminates a category of extraction error: nulling unknown fields prevents hallucination, confidence scores enable downstream quality filtering, and numeric-only amounts avoid currency symbol confusion.

You won't see a change yet. The next chunk adds the functions that use this prompt.

Write the extraction function

The extraction function is where your schema, prompt, and image come together. It encodes the invoice as base64, packages everything into a multimodal API request, and returns a fully typed InvoiceData object.

  • Below the EXTRACTION_PROMPT in extractor.py, add both extraction functions:
def encode_image(image_path: str) -> str:
    # Read the image file as bytes and encode to base64 string
    with open(image_path, "rb") as f:
        return base64.b64encode(f.read()).decode("utf-8")


def extract_invoice(image_path: str, client: OpenAI) -> InvoiceData:
    # Encode the image for the API
    base64_image = encode_image(image_path)

    # Detect media type from file extension
    file_ext = Path(image_path).suffix.lower()
    media_type = "image/png" if file_ext == ".png" else "image/jpeg"

    # Send multimodal request with Pydantic schema enforcement
    response = client.responses.parse(
        model="gpt-4o-mini",
        input=[
            {"role": "system", "content": EXTRACTION_PROMPT},
            {
                "role": "user",
                "content": [
                    {"type": "input_text", "text": "Extract all invoice data from this image."},
                    {
                        "type": "input_image",
                        "image_url": f"data:{media_type};base64,{base64_image}",
                    },
                ],
            },
        ],
        text_format=InvoiceData,
    )

    return response.output_parsed

What does this code do?

  • encode_image() reads any image file and converts it to a base64 string that the API can accept inline.
  • extract_invoice() detects whether the file is PNG or JPEG, then builds a multimodal request with both the system prompt and the image.
  • client.responses.parse() with text_format=InvoiceData tells OpenAI to constrain the model's output to your exact Pydantic schema. The response comes back already parsed into an InvoiceData object.

Why vision-first over OCR then text?

Sending the raw image directly to a vision model achieves roughly 92.7% accuracy on scanned invoices. The alternative (running OCR first, then passing extracted text to the model) drops to around 64.0% because OCR loses layout context like table structure and column alignment.

You won't see a change yet. The next chunk adds the CLI runner so you can test this.

Run the extractor against a sample invoice

To test the extractor from the command line, you need a __main__ block that accepts an image path as an argument and prints the extracted JSON.

  • At the bottom of extractor.py, add the CLI runner:
if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: python extractor.py <image_path>")
        sys.exit(1)

    client = OpenAI()
    result = extract_invoice(sys.argv[1], client)
    print(result.model_dump_json(indent=2))

What does this code do?

  • sys.argv[1] grabs the image path you pass on the command line.
  • OpenAI() creates a client that automatically reads your OPENAI_API_KEY environment variable.
  • model_dump_json(indent=2) converts the Pydantic object into pretty-printed JSON so you can inspect every extracted field.
  • Save extractor.py.
  • Run the extractor against one of your sample invoices by running this command:
python extractor.py invoices/sample1.png

You should see structured JSON printed to your terminal with fields like vendor_name, line_items, and total_amount populated from your invoice image. Each field will have an accompanying confidence score between 0.0 and 1.0.

Adjust the filename to match your image

If your sample invoice has a different filename or extension, replace invoices/sample1.png with the actual path. The extractor supports both PNG and JPEG formats.

Seeing an error?

  • AuthenticationError means your OPENAI_API_KEY environment variable is not set or has an invalid key. Re-export it in your terminal.
  • FileNotFoundError means the image path is wrong. Double-check the filename and that you're running the command from your project folder.
  • If you see ModuleNotFoundError: No module named 'openai', make sure your virtual environment is activated.

Help me debug my extractor error

✔️ Awesome, I've got everything!

Great. Double check you've saved your file before moving on.

ⓧ I'd like to double check the full code

import base64
import sys
from pathlib import Path

from openai import OpenAI
from schemas import InvoiceData


EXTRACTION_PROMPT = """You are a document extraction specialist. Extract all invoice data from this image.

Rules:
- Extract only text values visible in the document.
- If a field is not found, set it to null.
- For each field, provide a confidence score between 0.0 and 1.0.
- For line items, extract description, quantity, unit_price, and amount.
- Amounts should be numeric values without currency symbols.
- Dates should be in the format found in the document (do not normalize).
"""


def encode_image(image_path: str) -> str:
    with open(image_path, "rb") as f:
        return base64.b64encode(f.read()).decode("utf-8")


def extract_invoice(image_path: str, client: OpenAI) -> InvoiceData:
    base64_image = encode_image(image_path)

    file_ext = Path(image_path).suffix.lower()
    media_type = "image/png" if file_ext == ".png" else "image/jpeg"

    response = client.responses.parse(
        model="gpt-4o-mini",
        input=[
            {"role": "system", "content": EXTRACTION_PROMPT},
            {
                "role": "user",
                "content": [
                    {"type": "input_text", "text": "Extract all invoice data from this image."},
                    {
                        "type": "input_image",
                        "image_url": f"data:{media_type};base64,{base64_image}",
                    },
                ],
            },
        ],
        text_format=InvoiceData,
    )

    return response.output_parsed


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: python extractor.py <image_path>")
        sys.exit(1)

    client = OpenAI()
    result = extract_invoice(sys.argv[1], client)
    print(result.model_dump_json(indent=2))

Your extractor is working. Next up, you'll build the validation layer that catches extraction errors the model might make.

Add Deterministic Validation

Your extractor can pull structured data from invoice images, but LLMs sometimes misread digits, swap values, or hallucinate fields that don't exist.

Deterministic validation catches these errors with hard math checks and field rules that don't rely on the model's judgment. If a line items total doesn't add up to the subtotal, that's a definitive failure no matter how confident the model claims to be.

In this step, get ready to:
  • Build math checks that verify line items sum to subtotal and subtotal plus tax equals total.
  • Add field presence checks and confidence scoring thresholds.
  • Wire up all checks into a single validation orchestrator and test it against your extraction output.
Build the math validation rules

Invoices follow strict arithmetic rules. Line item amounts must sum to the subtotal, and subtotal plus tax must equal the total. If the model misreads a digit, these checks will catch it immediately.

  • Open validator.py in Cursor.
  • Add the import and the line items sum validation function by pasting this code:
from schemas import InvoiceData, ValidationCheck, ValidationResult


def validate_line_items_sum(invoice: InvoiceData) -> ValidationCheck:
    # If line items or subtotal are missing, we can't verify
    if not invoice.line_items or invoice.subtotal is None:
        return ValidationCheck(
            name="line_items_sum",
            status="warning",
            message="Cannot verify: missing line items or subtotal",
        )

    # Sum all line item amounts, skipping any that are None
    calculated_sum = sum(
        item.amount for item in invoice.line_items if item.amount is not None
    )
    # Allow a small rounding tolerance for floating point math
    tolerance = 0.02

    if abs(calculated_sum - invoice.subtotal) <= tolerance:
        return ValidationCheck(
            name="line_items_sum",
            status="pass",
            message=f"Line items sum ({calculated_sum:.2f}) matches subtotal ({invoice.subtotal:.2f})",
        )
    else:
        return ValidationCheck(
            name="line_items_sum",
            status="fail",
            message=f"Line items sum ({calculated_sum:.2f}) does not match subtotal ({invoice.subtotal:.2f})",
        )

What does this code do?

This function sums every line item's amount field and compares it to the extracted subtotal. A tolerance of 0.02 handles floating point rounding differences.

The three-state return (pass, fail, warning) is intentional. A "warning" means we can't even run the check because data is missing. A "fail" means the math is provably wrong.

  • Save validator.py.

You won't see output yet. The next function builds on this one.

  • Add the total validation function below your validate_line_items_sum function:
def validate_total(invoice: InvoiceData) -> ValidationCheck:
    # Need both subtotal and total to run this check
    if invoice.subtotal is None or invoice.total_amount is None:
        return ValidationCheck(
            name="total_check",
            status="warning",
            message="Cannot verify: missing subtotal or total",
        )

    # If tax is not extracted, assume zero
    tax = invoice.tax_amount if invoice.tax_amount is not None else 0.0
    expected_total = invoice.subtotal + tax
    tolerance = 0.02

    if abs(expected_total - invoice.total_amount) <= tolerance:
        return ValidationCheck(
            name="total_check",
            status="pass",
            message=f"Subtotal ({invoice.subtotal:.2f}) + tax ({tax:.2f}) = total ({invoice.total_amount:.2f})",
        )
    else:
        return ValidationCheck(
            name="total_check",
            status="fail",
            message=f"Subtotal ({invoice.subtotal:.2f}) + tax ({tax:.2f}) = {expected_total:.2f}, but total is {invoice.total_amount:.2f}",
        )

What does this code do?

This check verifies that subtotal + tax = total. If tax is missing from the extraction, it assumes zero rather than skipping the check entirely.

This is the most common extraction error on real invoices. Models frequently confuse subtotal and total values, or misread a tax line. This deterministic check catches that instantly.

  • Save validator.py.

Seeing a syntax error?

Make sure your validate_total function is at the same indentation level as validate_line_items_sum. Both should be top-level functions, not nested inside each other.

Help me fix my validator.py syntax error

Add field presence and confidence checks

Math checks catch arithmetic errors, but they can't catch a completely missing vendor name or a date the model guessed at with low confidence. These two checks handle the non-math validation.

  • Add the required fields validation below your validate_total function:
def validate_required_fields(invoice: InvoiceData) -> ValidationCheck:
    # These fields must be present for a usable extraction
    required = ["vendor_name", "invoice_number", "invoice_date", "total_amount"]
    missing = [f for f in required if getattr(invoice, f) is None]

    if not missing:
        return ValidationCheck(
            name="required_fields",
            status="pass",
            message="All required fields present",
        )
    else:
        return ValidationCheck(
            name="required_fields",
            status="fail",
            message=f"Missing required fields: {', '.join(missing)}",
        )

What does this code do?

This checks four critical fields that any valid invoice extraction must have: vendor name, invoice number, date, and total amount. If any are None, the extraction is incomplete.

The getattr(invoice, f) pattern lets you check field values dynamically using the field name as a string.

  • Save validator.py.
  • Add the confidence check below your validate_required_fields function:
def validate_confidence(invoice: InvoiceData) -> ValidationCheck:
    # Check confidence scores for key fields
    confidence_fields = [
        ("vendor_name", invoice.vendor_name_confidence),
        ("invoice_number", invoice.invoice_number_confidence),
        ("invoice_date", invoice.invoice_date_confidence),
        ("total_amount", invoice.total_amount_confidence),
    ]
    # Flag any field with confidence below 85%
    low_confidence = [
        name for name, conf in confidence_fields if conf is not None and conf < 0.85
    ]

    if not low_confidence:
        return ValidationCheck(
            name="confidence_check",
            status="pass",
            message="All key fields have high confidence",
        )
    else:
        return ValidationCheck(
            name="confidence_check",
            status="warning",
            message=f"Low confidence on: {', '.join(low_confidence)}",
        )

Why a warning instead of a fail?

Low confidence doesn't mean the extraction is wrong. It means the model is less certain. This surfaces fields that might need a second look without blocking the entire extraction.

The 0.85 threshold is a practical sweet spot. Lower values generate too many false alarms. Higher values miss genuinely uncertain extractions.

  • Save validator.py.

Not sure where to add each function?

Each function should be at the top level of the file (no indentation before def). They should appear one after another, separated by a single blank line.

Help me check my validator.py structure

Wire up the validation orchestrator and test it

Each individual check function handles one concern. Now you need an orchestrator that runs all four checks and produces a single ValidationResult with a clear pass/fail verdict and a list of flagged fields.

  • Add the validate_invoice orchestrator function below your validate_confidence function:
def validate_invoice(invoice: InvoiceData) -> ValidationResult:
    # Run all validation checks
    checks = [
        validate_line_items_sum(invoice),
        validate_total(invoice),
        validate_required_fields(invoice),
        validate_confidence(invoice),
    ]

    flagged_fields = []
    is_valid = True

    for check in checks:
        if check.status == "fail":
            is_valid = False
            # Map check failures to specific field names
            if "line_items" in check.name:
                flagged_fields.append("subtotal")
            elif "total" in check.name:
                flagged_fields.append("total_amount")
            elif "required" in check.name:
                flagged_fields.extend(
                    check.message.replace("Missing required fields: ", "").split(", ")
                )
        elif check.status == "warning" and "confidence" in check.name:
            # Low confidence fields get flagged but don't fail validation
            low_fields = check.message.replace("Low confidence on: ", "").split(", ")
            flagged_fields.extend(low_fields)

    return ValidationResult(
        checks=checks,
        is_valid=is_valid,
        flagged_fields=flagged_fields,
    )

What does this code do?

The orchestrator runs all four checks and collects results. If any check returns "fail", the entire validation is marked invalid. The flagged_fields list tracks exactly which fields are problematic.

This list is critical for the QA agent you'll build next. Instead of re-extracting the entire invoice, the agent uses flagged_fields to target only the specific values that need correction.

  • Save validator.py.
  • Test your validator against a real extraction by running this command in your terminal:
python -c "
from extractor import extract_invoice
from validator import validate_invoice
from openai import OpenAI

client = OpenAI()
invoice = extract_invoice('invoices/sample1.png', client)
result = validate_invoice(invoice)

print('Valid:', result.is_valid)
print('Flagged:', result.flagged_fields)
for check in result.checks:
    print(f'  [{check.status}] {check.name}: {check.message}')
"

You should see output listing each check with its status (pass, fail, or warning) and a human-readable message explaining the result.

What should I expect to see?

Your output will vary depending on the sample invoice. A well-structured invoice might pass all checks. A complex one might trigger a math mismatch or low confidence warning.

Both outcomes are useful. Passing checks confirm the extractor works. Failing checks prove the validator catches real errors.

Seeing an ImportError?

Make sure you're running the command from your project folder (the one containing validator.py, extractor.py, and schemas.py). Also confirm your virtual environment is active.

Help me fix my ImportError when running the validator

✔️ Awesome, I've got everything!

Great. Double check you've saved validator.py and move on to the next step.

ⓧ I'd like to double check the full code

Here's the complete validator.py file for comparison:

from schemas import InvoiceData, ValidationCheck, ValidationResult


def validate_line_items_sum(invoice: InvoiceData) -> ValidationCheck:
    if not invoice.line_items or invoice.subtotal is None:
        return ValidationCheck(
            name="line_items_sum",
            status="warning",
            message="Cannot verify: missing line items or subtotal",
        )

    calculated_sum = sum(
        item.amount for item in invoice.line_items if item.amount is not None
    )
    tolerance = 0.02

    if abs(calculated_sum - invoice.subtotal) <= tolerance:
        return ValidationCheck(
            name="line_items_sum",
            status="pass",
            message=f"Line items sum ({calculated_sum:.2f}) matches subtotal ({invoice.subtotal:.2f})",
        )
    else:
        return ValidationCheck(
            name="line_items_sum",
            status="fail",
            message=f"Line items sum ({calculated_sum:.2f}) does not match subtotal ({invoice.subtotal:.2f})",
        )


def validate_total(invoice: InvoiceData) -> ValidationCheck:
    if invoice.subtotal is None or invoice.total_amount is None:
        return ValidationCheck(
            name="total_check",
            status="warning",
            message="Cannot verify: missing subtotal or total",
        )

    tax = invoice.tax_amount if invoice.tax_amount is not None else 0.0
    expected_total = invoice.subtotal + tax
    tolerance = 0.02

    if abs(expected_total - invoice.total_amount) <= tolerance:
        return ValidationCheck(
            name="total_check",
            status="pass",
            message=f"Subtotal ({invoice.subtotal:.2f}) + tax ({tax:.2f}) = total ({invoice.total_amount:.2f})",
        )
    else:
        return ValidationCheck(
            name="total_check",
            status="fail",
            message=f"Subtotal ({invoice.subtotal:.2f}) + tax ({tax:.2f}) = {expected_total:.2f}, but total is {invoice.total_amount:.2f}",
        )


def validate_required_fields(invoice: InvoiceData) -> ValidationCheck:
    required = ["vendor_name", "invoice_number", "invoice_date", "total_amount"]
    missing = [f for f in required if getattr(invoice, f) is None]

    if not missing:
        return ValidationCheck(
            name="required_fields",
            status="pass",
            message="All required fields present",
        )
    else:
        return ValidationCheck(
            name="required_fields",
            status="fail",
            message=f"Missing required fields: {', '.join(missing)}",
        )


def validate_confidence(invoice: InvoiceData) -> ValidationCheck:
    confidence_fields = [
        ("vendor_name", invoice.vendor_name_confidence),
        ("invoice_number", invoice.invoice_number_confidence),
        ("invoice_date", invoice.invoice_date_confidence),
        ("total_amount", invoice.total_amount_confidence),
    ]
    low_confidence = [
        name for name, conf in confidence_fields if conf is not None and conf < 0.85
    ]

    if not low_confidence:
        return ValidationCheck(
            name="confidence_check",
            status="pass",
            message="All key fields have high confidence",
        )
    else:
        return ValidationCheck(
            name="confidence_check",
            status="warning",
            message=f"Low confidence on: {', '.join(low_confidence)}",
        )


def validate_invoice(invoice: InvoiceData) -> ValidationResult:
    checks = [
        validate_line_items_sum(invoice),
        validate_total(invoice),
        validate_required_fields(invoice),
        validate_confidence(invoice),
    ]

    flagged_fields = []
    is_valid = True

    for check in checks:
        if check.status == "fail":
            is_valid = False
            if "line_items" in check.name:
                flagged_fields.append("subtotal")
            elif "total" in check.name:
                flagged_fields.append("total_amount")
            elif "required" in check.name:
                flagged_fields.extend(
                    check.message.replace("Missing required fields: ", "").split(", ")
                )
        elif check.status == "warning" and "confidence" in check.name:
            low_fields = check.message.replace("Low confidence on: ", "").split(", ")
            flagged_fields.extend(low_fields)

    return ValidationResult(
        checks=checks,
        is_valid=is_valid,
        flagged_fields=flagged_fields,
    )

Your validation layer is complete. Next up, you'll build the QA agent that uses these check results to automatically correct extraction errors.

Build the QA Agent

Your extractor pulls structured data from invoices, and your validator catches errors like mismatched totals or missing fields. The problem is that validation only flags issues. It does not fix them.

In this step, you will build a QA agent that closes that gap. The agent takes validation failures, constructs a targeted correction prompt that tells the model exactly what went wrong, re-sends the image, and re-validates the output. This plan-act-verify loop is the same self-correction architecture used in production document processing systems.

In this step, get ready to:
  • Create a correction prompt template that feeds specific error messages back to the model.
  • Build a correction function that re-prompts the model with the original image and targeted fixes.
  • Wire up the full QA agent loop with a max-retries limit and overall confidence scoring.
Create the correction prompt and confidence scorer

The QA agent needs two things before it can start correcting: a prompt template that tells the model what it got wrong, and a scoring function that measures how confident the pipeline is in the final result.

  • Open qa_agent.py in Cursor and add the imports and correction prompt template:
import base64
from pathlib import Path

from openai import OpenAI
from schemas import ExtractionResult, InvoiceData, ValidationResult
from extractor import encode_image
from validator import validate_invoice


CORRECTION_PROMPT_TEMPLATE = """You are a document extraction specialist performing a quality review.

A previous extraction attempt produced errors. Here are the specific issues found:

{issues}

Please re-examine the invoice image carefully and provide a corrected extraction.
Pay special attention to the flagged fields. Common errors include:
- Misreading digits (e.g., 8 vs 6, 1 vs 7)
- Incorrect decimal placement
- Missing line items
- Swapping subtotal and total values

Rules:
- Extract only text values visible in the document.
- If a field is not found, set it to null.
- For each field, provide a confidence score between 0.0 and 1.0.
- Amounts should be numeric values without currency symbols.
"""

What does this code do?

The imports pull in your existing schemas, the image encoder from your extractor, and the validation function. You are reusing these building blocks rather than rewriting them.

The CORRECTION_PROMPT_TEMPLATE is the key to self-correction. The {issues} placeholder gets filled at runtime with the exact validation failures. This gives the model diagnostic information about what it got wrong so it can target those specific fields on the retry.

The confidence scorer combines two signals into one number: how confident the model is in its own extractions, and how many deterministic validation checks passed.

  • Below the CORRECTION_PROMPT_TEMPLATE string, add the confidence scoring function:
def calculate_overall_confidence(invoice: InvoiceData, validation: ValidationResult) -> float:
    confidence_values = []

    # Collect all field-level confidence scores reported by the model
    field_confidences = [
        invoice.vendor_name_confidence,
        invoice.invoice_number_confidence,
        invoice.invoice_date_confidence,
        invoice.subtotal_confidence,
        invoice.tax_amount_confidence,
        invoice.total_amount_confidence,
    ]
    for conf in field_confidences:
        if conf is not None:
            confidence_values.append(conf)

    # Default to 0.5 if no confidence values were reported
    if not confidence_values:
        field_avg = 0.5
    else:
        field_avg = sum(confidence_values) / len(confidence_values)

    # Calculate what percentage of validation checks passed
    total_checks = len(validation.checks)
    passed_checks = sum(1 for c in validation.checks if c.status == "pass")
    validation_score = passed_checks / total_checks if total_checks > 0 else 0.5

    # Weighted combination: 60% model confidence, 40% validation pass rate
    overall = (field_avg * 0.6) + (validation_score * 0.4)
    return round(overall, 3)

Why two signals instead of one?

The model's self-reported confidence tells you how sure it is about what it read. The validation score tells you whether the numbers actually add up. A model might report 0.95 confidence on a total that fails the math check. Combining both catches cases where either signal alone would mislead you.

The 60/40 weighting prioritizes field confidence because it covers more ground (six fields vs four checks). The final score determines whether a document is auto-approved or routed to human review.

  • Save qa_agent.py.

You will not see output yet. The next two chunks use these building blocks to actually run corrections.

Seeing a syntax error on save?

Check that the triple-quoted string for CORRECTION_PROMPT_TEMPLATE has matching """ at both the start and end. A missing closing set of triple quotes causes Python to swallow everything below it into the string.

Also verify that {issues} uses curly braces, not square brackets.

Help me debug a syntax error in my qa_agent.py correction prompt template.

Build the correction function

The correction function is where the agent acts on its diagnosis. It takes the validation results, formats the specific failures into the correction prompt, and sends the original image back to the model with instructions to fix those exact issues.

  • Below calculate_overall_confidence, add the attempt_correction function:
def attempt_correction(
    image_path: str,
    validation: ValidationResult,
    client: OpenAI,
) -> InvoiceData:
    # Format failed checks into a bullet list for the correction prompt
    issues = "\n".join(
        f"- [{check.name}] {check.message}"
        for check in validation.checks
        if check.status == "fail"
    )

    base64_image = encode_image(image_path)
    file_ext = Path(image_path).suffix.lower()
    media_type = "image/png" if file_ext == ".png" else "image/jpeg"

    # Insert the specific issues into the correction prompt template
    prompt = CORRECTION_PROMPT_TEMPLATE.format(issues=issues)

    response = client.responses.parse(
        model="gpt-4o-mini",
        input=[
            {"role": "system", "content": prompt},
            {
                "role": "user",
                "content": [
                    {"type": "input_text", "text": "Re-extract the invoice data, correcting the issues listed above."},
                    {
                        "type": "input_image",
                        "image_url": f"data:{media_type};base64,{base64_image}",
                    },
                ],
            },
        ],
        text_format=InvoiceData,
    )

    return response.output_parsed

What does this code do?

  • The issues variable filters validation checks to only those that failed, then formats them as a readable bullet list. This is the diagnostic feedback the model receives.
  • The function re-encodes the original image and sends it again. The model gets a fresh look at the document, but this time with explicit guidance about what went wrong.
  • The response uses the same text_format=InvoiceData pattern as the initial extraction. Structured Outputs guarantees the corrected result still matches your schema.

Why is this different from a simple retry?

A naive retry sends the exact same prompt and hopes for better luck. This correction function sends a different prompt that includes the specific error messages. The model knows it previously produced a subtotal that does not match line items, or a total that fails the math check. This targeted feedback is what makes it a QA agent rather than a retry loop.

  • Save qa_agent.py.

You still will not see output. The next chunk wires this into the full agent loop.

Seeing an IndentationError?

The function signature spans multiple lines with each parameter on its own line. Make sure the closing parenthesis of the signature is followed by -> InvoiceData: on the same line. The function body must be indented one level (4 spaces) from the def keyword.

Help me fix an indentation error in my attempt_correction function.

Wire up the QA agent loop

The agent loop is the orchestrator. It validates the initial extraction, and if issues are found, it calls the correction function, re-validates, and repeats until the output passes or the retry limit is reached.

  • Below attempt_correction, add the run_qa_agent function:
def run_qa_agent(
    image_path: str,
    initial_extraction: InvoiceData,
    client: OpenAI,
    max_retries: int = 2,
) -> ExtractionResult:
    current_extraction = initial_extraction
    validation = validate_invoice(current_extraction)
    attempts = 1
    corrections_made = []

    # Loop until validation passes or we hit the retry limit
    while not validation.is_valid and attempts <= max_retries:
        attempts += 1
        corrections_made.append(
            f"Attempt {attempts}: Correcting {', '.join(validation.flagged_fields)}"
        )

        # Re-extract with targeted correction prompt
        current_extraction = attempt_correction(image_path, validation, client)
        validation = validate_invoice(current_extraction)

    # Score the final result regardless of whether all checks passed
    overall_confidence = calculate_overall_confidence(current_extraction, validation)

    return ExtractionResult(
        invoice=current_extraction,
        validation=validation,
        attempts=attempts,
        corrections_made=corrections_made,
        overall_confidence=overall_confidence,
    )

What does this code do?

  • The while loop runs only if validation failed AND the agent has retries remaining. A perfectly extracted invoice skips the loop entirely.
  • Each iteration logs which fields are being corrected in the corrections_made list. This gives you an audit trail of what the agent did.
  • The max_retries=2 default means the agent makes at most 2 correction attempts after the initial extraction. This prevents infinite loops and keeps API costs predictable.
  • The final ExtractionResult bundles everything: the invoice data, validation status, how many attempts were needed, what was corrected, and the overall confidence score.
  • Save qa_agent.py.
  • Verify your file has no import errors by running:
python -c "from qa_agent import run_qa_agent; print('QA agent module loaded successfully')"

You should see QA agent module loaded successfully printed in your terminal.

Seeing an ImportError?

If the error mentions cannot import name 'encode_image', make sure your extractor.py file is in the same folder as qa_agent.py and that the encode_image function exists in it.

If the error mentions cannot import name 'ExtractionResult', check that your schemas.py includes the ExtractionResult class.

Help me debug an ImportError in my qa_agent.py file.

✔️ Awesome, I've got everything!

Great. Double check you have saved qa_agent.py and the import test passed.

ⓧ I'd like to double check the full code

Here is the complete qa_agent.py file:

import base64
from pathlib import Path

from openai import OpenAI
from schemas import ExtractionResult, InvoiceData, ValidationResult
from extractor import encode_image
from validator import validate_invoice


CORRECTION_PROMPT_TEMPLATE = """You are a document extraction specialist performing a quality review.

A previous extraction attempt produced errors. Here are the specific issues found:

{issues}

Please re-examine the invoice image carefully and provide a corrected extraction.
Pay special attention to the flagged fields. Common errors include:
- Misreading digits (e.g., 8 vs 6, 1 vs 7)
- Incorrect decimal placement
- Missing line items
- Swapping subtotal and total values

Rules:
- Extract only text values visible in the document.
- If a field is not found, set it to null.
- For each field, provide a confidence score between 0.0 and 1.0.
- Amounts should be numeric values without currency symbols.
"""


def calculate_overall_confidence(invoice: InvoiceData, validation: ValidationResult) -> float:
    confidence_values = []

    field_confidences = [
        invoice.vendor_name_confidence,
        invoice.invoice_number_confidence,
        invoice.invoice_date_confidence,
        invoice.subtotal_confidence,
        invoice.tax_amount_confidence,
        invoice.total_amount_confidence,
    ]
    for conf in field_confidences:
        if conf is not None:
            confidence_values.append(conf)

    if not confidence_values:
        field_avg = 0.5
    else:
        field_avg = sum(confidence_values) / len(confidence_values)

    total_checks = len(validation.checks)
    passed_checks = sum(1 for c in validation.checks if c.status == "pass")
    validation_score = passed_checks / total_checks if total_checks > 0 else 0.5

    overall = (field_avg * 0.6) + (validation_score * 0.4)
    return round(overall, 3)


def attempt_correction(
    image_path: str,
    validation: ValidationResult,
    client: OpenAI,
) -> InvoiceData:
    issues = "\n".join(
        f"- [{check.name}] {check.message}"
        for check in validation.checks
        if check.status == "fail"
    )

    base64_image = encode_image(image_path)
    file_ext = Path(image_path).suffix.lower()
    media_type = "image/png" if file_ext == ".png" else "image/jpeg"

    prompt = CORRECTION_PROMPT_TEMPLATE.format(issues=issues)

    response = client.responses.parse(
        model="gpt-4o-mini",
        input=[
            {"role": "system", "content": prompt},
            {
                "role": "user",
                "content": [
                    {"type": "input_text", "text": "Re-extract the invoice data, correcting the issues listed above."},
                    {
                        "type": "input_image",
                        "image_url": f"data:{media_type};base64,{base64_image}",
                    },
                ],
            },
        ],
        text_format=InvoiceData,
    )

    return response.output_parsed


def run_qa_agent(
    image_path: str,
    initial_extraction: InvoiceData,
    client: OpenAI,
    max_retries: int = 2,
) -> ExtractionResult:
    current_extraction = initial_extraction
    validation = validate_invoice(current_extraction)
    attempts = 1
    corrections_made = []

    while not validation.is_valid and attempts <= max_retries:
        attempts += 1
        corrections_made.append(
            f"Attempt {attempts}: Correcting {', '.join(validation.flagged_fields)}"
        )

        current_extraction = attempt_correction(image_path, validation, client)
        validation = validate_invoice(current_extraction)

    overall_confidence = calculate_overall_confidence(current_extraction, validation)

    return ExtractionResult(
        invoice=current_extraction,
        validation=validation,
        attempts=attempts,
        corrections_made=corrections_made,
        overall_confidence=overall_confidence,
    )

Your QA agent is ready. It can diagnose extraction failures, construct targeted correction prompts, and retry with a confidence-based stopping criterion. Next up, you will wire everything together into an end-to-end pipeline that processes invoices and prints a full report.

Wire Up the End-to-End Pipeline

You now have all four building blocks ready: schemas, extractor, validator, and QA agent. Each works independently, but the real power comes from connecting them into a single automated flow.

In this step, you'll build main.py to orchestrate the full pipeline. One command will extract data from an invoice image, validate the results, trigger QA corrections if needed, and print a formatted report with confidence scores and an approval status.

In this step, get ready to:
  • Write a report formatter that displays extraction results, validation checks, and QA metrics.
  • Build the pipeline orchestrator that chains extraction, validation, and self-correction.
  • Run the full pipeline against your sample invoices and review per-field confidence scores.
Write the report formatter

Before building the orchestration logic, you need a way to display results. The report formatter takes an ExtractionResult and prints a structured summary showing extracted data, validation check results, and the QA agent's activity.

  • Open main.py in Cursor and add the imports and report formatter by pasting this code:
import sys
from pathlib import Path

from openai import OpenAI
from schemas import ExtractionResult
from extractor import extract_invoice
from validator import validate_invoice
from qa_agent import run_qa_agent, calculate_overall_confidence


def print_report(result: ExtractionResult, image_path: str) -> None:
    print(f"\n{'='*60}")
    print(f"EXTRACTION REPORT: {Path(image_path).name}")
    print(f"{'='*60}")

    invoice = result.invoice
    print(f"\n--- Extracted Data ---")
    print(f"Vendor:         {invoice.vendor_name or 'N/A'}")
    print(f"Invoice #:      {invoice.invoice_number or 'N/A'}")
    print(f"Date:           {invoice.invoice_date or 'N/A'}")
    print(f"Customer:       {invoice.customer_name or 'N/A'}")

    if invoice.line_items:
        print(f"\n--- Line Items ({len(invoice.line_items)}) ---")
        for i, item in enumerate(invoice.line_items, 1):
            qty = f"x{item.quantity}" if item.quantity else ""
            price = f"@ ${item.unit_price:.2f}" if item.unit_price else ""
            amount = f"= ${item.amount:.2f}" if item.amount else ""
            print(f"  {i}. {item.description} {qty} {price} {amount}")

    print(f"\n--- Totals ---")
    print(f"Subtotal:       ${invoice.subtotal:.2f}" if invoice.subtotal else "Subtotal:       N/A")
    print(f"Tax:            ${invoice.tax_amount:.2f}" if invoice.tax_amount else "Tax:            N/A")
    print(f"Total:          ${invoice.total_amount:.2f}" if invoice.total_amount else "Total:          N/A")

    print(f"\n--- Validation ---")
    for check in result.validation.checks:
        icon = "\u2705" if check.status == "pass" else "\u274c" if check.status == "fail" else "\u26a0\ufe0f"
        print(f"  {icon} {check.name}: {check.message}")

    print(f"\n--- QA Summary ---")
    print(f"Attempts:           {result.attempts}")
    print(f"Corrections Made:   {len(result.corrections_made)}")
    for correction in result.corrections_made:
        print(f"    - {correction}")
    print(f"Overall Confidence: {result.overall_confidence:.1%}")
    print(f"Status:             {'APPROVED' if result.overall_confidence >= 0.85 else 'NEEDS REVIEW'}")
    print(f"{'='*60}\n")

What does this code do?

The imports pull in all four modules you built in previous steps. The function itself prints a human-readable report with four sections:

  • Extracted Data shows the key fields the model pulled from the invoice (vendor, invoice number, date, customer).
  • Line Items lists each item with quantity, unit price, and total amount.
  • Validation shows each check result with a color-coded emoji icon.
  • QA Summary reports how many attempts were made, what was corrected, and whether the final confidence score meets the 0.85 threshold for automatic approval.
  • Save main.py.

Seeing an import error?

Make sure all four files (schemas.py, extractor.py, validator.py, qa_agent.py) are saved in the same folder as main.py. Python can only import from files in the same directory or on the path.

Help me debug my import error in main.py

Build the pipeline orchestrator

The process_invoice function is the heart of your system. It chains the three stages together: extract, validate, and conditionally correct.

  • In main.py, add the process_invoice function below print_report by pasting this code:
def process_invoice(image_path: str, client: OpenAI) -> ExtractionResult:
    print(f"\nProcessing: {image_path}")
    print("Step 1: Extracting data with vision model...")
    initial_extraction = extract_invoice(image_path, client)

    print("Step 2: Running validation checks...")
    validation = validate_invoice(initial_extraction)

    if validation.is_valid:
        print("Step 3: All checks passed! No QA correction needed.")
        overall = calculate_overall_confidence(initial_extraction, validation)
        result = ExtractionResult(
            invoice=initial_extraction,
            validation=validation,
            attempts=1,
            corrections_made=[],
            overall_confidence=overall,
        )
    else:
        print(f"Step 3: Issues found. Running QA agent (flagged: {', '.join(validation.flagged_fields)})...")
        result = run_qa_agent(image_path, initial_extraction, client)

    print_report(result, image_path)
    return result

What does this code do?

This function implements the plan-act-verify pattern that makes this a real QA agent rather than a simple retry loop:

  • extract_invoice() sends the image to GPT-4o-mini and gets back structured data.
  • validate_invoice() runs all four deterministic checks against the extraction.
  • If validation passes, it calculates confidence and returns immediately. If validation fails, it hands off to run_qa_agent() which re-prompts with specific error messages.
  • Save main.py.

Not sure where to paste this?

Place process_invoice directly after the closing line of print_report. Leave a blank line between the two functions. The order matters because process_invoice calls print_report at the end.

Help me figure out the correct order of functions in main.py

Add the CLI entry point and run the pipeline

The final piece is a main() function that reads image paths from the command line and feeds each one through the pipeline. This lets you process one invoice or a whole batch in a single command.

  • Add the main function and entry point below process_invoice by pasting this code:
def main():
    if len(sys.argv) < 2:
        print("Usage: python main.py <image_path> [image_path2 ...]")
        print("Example: python main.py invoices/sample1.png invoices/sample2.jpg")
        sys.exit(1)

    client = OpenAI()

    for image_path in sys.argv[1:]:
        if not Path(image_path).exists():
            print(f"Error: File not found: {image_path}")
            continue
        process_invoice(image_path, client)


if __name__ == "__main__":
    main()

What does this code do?

  • sys.argv[1:] reads all image paths passed after the script name on the command line.
  • The file existence check with Path(image_path).exists() skips missing files gracefully instead of crashing mid-batch.
  • A single OpenAI() client is created once and reused for all invoices, which avoids redundant authentication.
  • Save main.py.
  • Run the full pipeline against your sample invoices by running this command:
python main.py invoices/sample1.png invoices/sample2.png

What should I see?

You should see progress messages as each invoice is processed, followed by a formatted report for each one. The report includes:

  • Extracted vendor name, invoice number, date, and customer.
  • A numbered list of line items with quantities and amounts.
  • Validation check results with pass/fail/warning icons.
  • A QA summary showing attempts, corrections, overall confidence percentage, and an APPROVED or NEEDS REVIEW status.

If the QA agent was triggered, you will also see which fields were flagged and what it corrected on each retry attempt.

How do I interpret the confidence metrics?

The overall confidence score combines two signals: per-field confidence (weighted at 60%) and validation pass rate (weighted at 40%). Documents scoring 85% or above are marked APPROVED. Anything below goes into NEEDS REVIEW status, meaning a human should check it.

Look at which fields triggered corrections. If the same field (like subtotal or total_amount) keeps getting flagged across multiple invoices, that tells you the model struggles with that specific extraction task.

Seeing an API error or timeout?

Make sure your OPENAI_API_KEY environment variable is still set in your current terminal session. If you opened a new terminal window, you need to export it again.

If you see a rate limit error, wait 30 seconds and try again. Processing multiple images sends several API calls in quick succession.

Help me debug my API error when running main.py

✔️ Awesome, I've got everything!

Great. Double check you have saved main.py before moving on.

ⓧ I'd like to double check the full code

Here is the complete main.py file for reference:

import sys
from pathlib import Path

from openai import OpenAI
from schemas import ExtractionResult
from extractor import extract_invoice
from validator import validate_invoice
from qa_agent import run_qa_agent, calculate_overall_confidence


def print_report(result: ExtractionResult, image_path: str) -> None:
    print(f"\n{'='*60}")
    print(f"EXTRACTION REPORT: {Path(image_path).name}")
    print(f"{'='*60}")

    invoice = result.invoice
    print(f"\n--- Extracted Data ---")
    print(f"Vendor:         {invoice.vendor_name or 'N/A'}")
    print(f"Invoice #:      {invoice.invoice_number or 'N/A'}")
    print(f"Date:           {invoice.invoice_date or 'N/A'}")
    print(f"Customer:       {invoice.customer_name or 'N/A'}")

    if invoice.line_items:
        print(f"\n--- Line Items ({len(invoice.line_items)}) ---")
        for i, item in enumerate(invoice.line_items, 1):
            qty = f"x{item.quantity}" if item.quantity else ""
            price = f"@ ${item.unit_price:.2f}" if item.unit_price else ""
            amount = f"= ${item.amount:.2f}" if item.amount else ""
            print(f"  {i}. {item.description} {qty} {price} {amount}")

    print(f"\n--- Totals ---")
    print(f"Subtotal:       ${invoice.subtotal:.2f}" if invoice.subtotal else "Subtotal:       N/A")
    print(f"Tax:            ${invoice.tax_amount:.2f}" if invoice.tax_amount else "Tax:            N/A")
    print(f"Total:          ${invoice.total_amount:.2f}" if invoice.total_amount else "Total:          N/A")

    print(f"\n--- Validation ---")
    for check in result.validation.checks:
        icon = "\u2705" if check.status == "pass" else "\u274c" if check.status == "fail" else "\u26a0\ufe0f"
        print(f"  {icon} {check.name}: {check.message}")

    print(f"\n--- QA Summary ---")
    print(f"Attempts:           {result.attempts}")
    print(f"Corrections Made:   {len(result.corrections_made)}")
    for correction in result.corrections_made:
        print(f"    - {correction}")
    print(f"Overall Confidence: {result.overall_confidence:.1%}")
    print(f"Status:             {'APPROVED' if result.overall_confidence >= 0.85 else 'NEEDS REVIEW'}")
    print(f"{'='*60}\n")


def process_invoice(image_path: str, client: OpenAI) -> ExtractionResult:
    print(f"\nProcessing: {image_path}")
    print("Step 1: Extracting data with vision model...")
    initial_extraction = extract_invoice(image_path, client)

    print("Step 2: Running validation checks...")
    validation = validate_invoice(initial_extraction)

    if validation.is_valid:
        print("Step 3: All checks passed! No QA correction needed.")
        overall = calculate_overall_confidence(initial_extraction, validation)
        result = ExtractionResult(
            invoice=initial_extraction,
            validation=validation,
            attempts=1,
            corrections_made=[],
            overall_confidence=overall,
        )
    else:
        print(f"Step 3: Issues found. Running QA agent (flagged: {', '.join(validation.flagged_fields)})...")
        result = run_qa_agent(image_path, initial_extraction, client)

    print_report(result, image_path)
    return result


def main():
    if len(sys.argv) < 2:
        print("Usage: python main.py <image_path> [image_path2 ...]")
        print("Example: python main.py invoices/sample1.png invoices/sample2.jpg")
        sys.exit(1)

    client = OpenAI()

    for image_path in sys.argv[1:]:
        if not Path(image_path).exists():
            print(f"Error: File not found: {image_path}")
            continue
        process_invoice(image_path, client)


if __name__ == "__main__":
    main()

Your document extraction pipeline is fully operational. You have a system that extracts structured data from invoices, validates the results with deterministic checks, and automatically corrects its own mistakes. Next up, the Secret Mission challenges you to build a Streamlit UI and batch processor on top of this pipeline.

Secret mission

Add a Streamlit Demo UI and Batch Processor

Your pipeline works from the command line, but what if you could drag and drop an invoice into a web interface and watch the extraction happen in real time? In this secret mission, you will build a Streamlit demo UI and a batch processor that generates accuracy reports across multiple invoices.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing cloud costs. The only expense was the OpenAI API credits used during extraction (typically $0.05-0.15 total). Decide whether to keep your project files, pause your environment, or delete everything.

Resources you used:

  • Project folder with Python files (schemas.py, extractor.py, validator.py, qa_agent.py, main.py).
  • Python virtual environment.
  • OpenAI API key set as an environment variable.
  • Sample invoice images in your project folder.
  • Installed Python packages (openai, pydantic, streamlit).

Keep everything running

No action needed. Choose this if you plan to keep experimenting with the pipeline or processing more invoices.

  • Your project files stay on your Desktop.
  • Your virtual environment stays ready to activate.
  • Your OpenAI API key remains set in your shell configuration.
  • You will only incur further OpenAI costs if you run the extraction scripts again.

Pause - I'll come back to this later

Deactivate your virtual environment to free up your terminal session. Your files and packages stay intact for when you return.

  • Run deactivate in your terminal to exit the virtual environment.
  • When you come back, navigate to your project folder and run source venv/bin/activate to reactivate.
  • Your OpenAI API key stays configured in your shell profile.

Delete - I don't want to use this again

Remove all project files and the virtual environment from your system.

  • Deactivate your virtual environment by running deactivate.
  • Delete the entire project folder from your Desktop.
  • Optionally, remove the OPENAI_API_KEY line from your shell profile (~/.zshrc on macOS or ~/.bashrc on Linux) if you no longer need it.
  • If you want to revoke your OpenAI API key entirely, visit the API keys page in your OpenAI account settings and delete the key.

Nice Work!

Nice Work!

You just built a production-quality document extraction pipeline with a self-correcting QA agent.

This is the exact architecture that companies like Extend use in production: vision-based extraction, schema enforcement, deterministic validation, and an agentic self-correction loop.

You learned how to:

  • Build a vision-based document extractor using GPT-4o-mini and OpenAI Structured Outputs with Pydantic schemas for type-safe, validated extraction.
  • Implement deterministic validation rules including math reconciliation, format checks, and required field enforcement to catch extraction errors.
  • Create a QA agent that flags low-confidence results, constructs targeted correction prompts, and attempts automatic self-correction with confidence scoring.
  • Build a Streamlit demo UI for drag-and-drop invoice processing and a batch processor with aggregate accuracy reporting.

Ready to quiz yourself?