Automated invoice processing is rapidly becoming a foundational workflow in modern finance operations. In 2026, the integration of AI with human-in-the-loop (HITL) review steps is no longer a luxury—it's a necessity for accuracy, compliance, and scalability. As we covered in our complete guide to AI workflow automation for finance and accounting in 2026, invoice processing is a prime candidate for intelligent automation, but also demands careful design to handle edge cases and exceptions.
In this deep-dive tutorial, you'll learn how to build a robust, end-to-end AI workflow for invoice processing, with HITL steps for validation and correction. We'll use open-source tools and cloud APIs common in 2026, and provide all the code, configuration, and troubleshooting tips you need to get to production-readiness.
Prerequisites
- Python 3.11+ (with
pipandvenv) - Docker (v25+)
- PostgreSQL (v15+)
- OpenAI GPT-4o API access (or equivalent LLM API)
- Tesseract OCR (v5+)
- Experience with REST APIs and webhooks
- Basic knowledge of
FastAPI,Celery, andReact(for optional UI) - Familiarity with invoice data structures (PDF, image, UBL/XML, JSON)
- Sample invoice files (PDFs, scans, or images)
Overview of the Workflow
- Invoice Ingestion: Receive invoices via email, upload, or API.
- Document Preprocessing: Convert files to images/text using OCR.
- AI Extraction: Use LLMs to extract structured data.
- Validation: Auto-validate with business rules, flagging uncertain cases.
- Human-in-the-Loop Review: Present flagged invoices for manual review.
- Approval & Posting: Update records in the finance system.
- Audit Logging: Store all actions and corrections for compliance.
1. Set Up Your Development Environment
-
Create a Python Virtual Environment
python3 -m venv venv source venv/bin/activate
-
Install Required Packages
pip install fastapi uvicorn[standard] celery[redis] pydantic sqlalchemy psycopg2-binary openai pillow pytesseract python-dotenv
-
Install Tesseract OCR
brew install tesseract sudo apt-get install tesseract-ocr -
Start PostgreSQL (Docker example)
docker run --name invoices-db -e POSTGRES_PASSWORD=secret -e POSTGRES_DB=invoices -p 5432:5432 -d postgres:15 -
Set up .env file with API keys and DB connection
OPENAI_API_KEY=sk-... DATABASE_URL=postgresql://postgres:secret@localhost:5432/invoices
2. Invoice Ingestion: API Endpoint & File Handling
-
Create a FastAPI app with an upload endpoint
from fastapi import FastAPI, UploadFile, File, HTTPException import shutil import os app = FastAPI() UPLOAD_DIR = "uploads" os.makedirs(UPLOAD_DIR, exist_ok=True) @app.post("/invoices/upload") async def upload_invoice(file: UploadFile = File(...)): file_path = os.path.join(UPLOAD_DIR, file.filename) with open(file_path, "wb") as buffer: shutil.copyfileobj(file.file, buffer) return {"filename": file.filename, "path": file_path} -
Test the upload endpoint
uvicorn app.main:app --reload curl -F "file=@/path/to/invoice.pdf" http://localhost:8000/invoices/upload -
Optional: Set up email ingestion using IMAP or cloud mail API
For advanced setups, see our step-by-step AI invoice processing tutorial.
3. Document Preprocessing with OCR
-
Convert PDFs/images to text using Tesseract
from PIL import Image import pytesseract from pdf2image import convert_from_path def extract_text(file_path): if file_path.lower().endswith(".pdf"): images = convert_from_path(file_path) text = "" for img in images: text += pytesseract.image_to_string(img) return text else: img = Image.open(file_path) return pytesseract.image_to_string(img) -
Install PDF dependencies
pip install pdf2image
sudo apt-get install poppler-utils -
Test OCR extraction
from app.ocr import extract_text print(extract_text("uploads/sample_invoice.pdf"))
4. AI Extraction: Structured Data from Text
-
Define the extraction prompt for LLM
import openai import os def extract_invoice_data(text): prompt = f""" Extract the following fields from the invoice text below: - Invoice Number - Invoice Date - Vendor Name - Total Amount - Line Items (description, quantity, unit price, total) Reply in JSON format. Invoice Text: {text} """ response = openai.ChatCompletion.create( model="gpt-4o", messages=[{"role": "user", "content": prompt}], temperature=0.0, max_tokens=1024 ) return response.choices[0].message['content'] -
Test LLM extraction
from app.ocr import extract_text from app.extract import extract_invoice_data text = extract_text("uploads/sample_invoice.pdf") structured = extract_invoice_data(text) print(structured)For advanced prompt techniques, see Prompt Engineering for Financial Reporting Automation: The 2026 Playbook.
5. Validation and Auto-Flagging for Human Review
-
Implement business rule validation
import json from datetime import datetime def validate_invoice(data_str): data = json.loads(data_str) flags = [] # Example rules try: invoice_date = datetime.strptime(data['Invoice Date'], "%Y-%m-%d") if invoice_date > datetime.now(): flags.append("Invoice date is in the future.") except Exception: flags.append("Invalid invoice date format.") try: total = float(data['Total Amount'].replace("$", "")) if total <= 0: flags.append("Total amount is zero or negative.") except Exception: flags.append("Invalid total amount.") # Add more rules as needed return flags -
Auto-flag invoices for HITL review
flags = validate_invoice(structured) if flags: print("Invoice requires human review:", flags) else: print("Invoice auto-approved.") -
Store flagged invoices in the database
from sqlalchemy import Column, Integer, String, Boolean, JSON, create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker Base = declarative_base() class Invoice(Base): __tablename__ = "invoices" id = Column(Integer, primary_key=True) filename = Column(String) data = Column(JSON) requires_review = Column(Boolean, default=False) flags = Column(JSON) engine = create_engine(os.getenv("DATABASE_URL")) Base.metadata.create_all(engine) Session = sessionmaker(bind=engine)
6. Human-in-the-Loop Review Interface
-
Expose a REST API for flagged invoices
from fastapi import Depends from sqlalchemy.orm import Session from app.models import Invoice, Session as DBSession @app.get("/invoices/flagged") def get_flagged_invoices(): session = DBSession() invoices = session.query(Invoice).filter(Invoice.requires_review == True).all() return [{"id": inv.id, "filename": inv.filename, "data": inv.data, "flags": inv.flags} for inv in invoices] @app.post("/invoices/review/{invoice_id}") def review_invoice(invoice_id: int, corrected_data: dict): session = DBSession() inv = session.query(Invoice).get(invoice_id) inv.data = corrected_data inv.requires_review = False session.commit() return {"status": "reviewed"} -
Build a minimal React UI (optional)
For each flagged invoice, display extracted data, flags, and allow correction/approval. (See screenshot description below.)
- Screenshot Description: The UI shows a table of flagged invoices. Clicking an invoice opens a form with fields (Invoice Number, Date, Vendor, etc.), highlighted errors, and buttons to "Approve" or "Edit & Save".
For detailed HITL patterns, see How to Build Human-in-the-Loop Review Steps in Automated Customer Service Workflows.
7. Approval, Posting, and Audit Logging
-
On approval, post data to your finance system (API or DB)
def post_to_finance_system(invoice_data): # Example: POST to ERP API or insert into DB pass # Replace with integration code -
Log all actions for compliance
from datetime import datetime def log_action(invoice_id, action, user, details): # Write to audit table or log file with open("audit.log", "a") as f: f.write(f"{datetime.now()} | Invoice {invoice_id} | {action} | {user} | {details}\n")
8. Orchestrate the Workflow with Celery
-
Set up a Celery worker for asynchronous processing
from celery import Celery celery_app = Celery("invoice_tasks", broker="redis://localhost:6379/0") @celery_app.task def process_invoice(file_path): text = extract_text(file_path) data = extract_invoice_data(text) flags = validate_invoice(data) # Save to DB, trigger HITL if needed -
Start the Celery worker
celery -A app.tasks worker --loglevel=info -
Trigger processing after upload
from app.tasks import process_invoice process_invoice.delay(file_path)
Common Issues & Troubleshooting
- OCR returns poor quality text: Ensure invoices are high-resolution; try different Tesseract language/data models.
- LLM extraction errors: Refine prompts, limit input to relevant text, or use LLM function-calling APIs for structured output.
-
Database connection fails: Check
DATABASE_URL, ensure PostgreSQL is running and accessible. -
Celery tasks not running: Verify Redis is running (
docker run -p 6379:6379 redis
) and broker URL matches. - API authentication: For production, secure endpoints with OAuth2/JWT.
-
Human review UI not updating: Confirm backend review endpoint updates
requires_reviewflag and UI polls for status. - LLM costs/latency: Batch requests, use lower-cost models for initial parsing, or self-host open-source alternatives.
Next Steps
Congratulations! You now have a working AI workflow for automated invoice processing with human-in-the-loop review—ready for scaling, compliance, and integration with your finance stack. To further enhance your solution:
- Implement advanced business rules and duplicate detection (see AI Workflow Automation for Finance: 2026’s Most Common Mistakes (and How to Avoid Them)).
- Integrate real-time fraud detection (see AI Workflow Automation for Real-Time Fraud Detection: Visa’s 2026 Rollout).
- Expand to other finance workflows, like reconciliation (see Workflow Automation for Real-Time Financial Reconciliation: AI-Powered Best Practices (2026)).
- Compare workflow automation tools and platforms (see Best AI Workflow Automation Tools for Finance Teams in 2026).
For a broader strategy and technology landscape, revisit our pillar guide to mastering AI workflow automation for finance & accounting in 2026.