What is an Invoice Processing Agent?
An Invoice Processing Agent is an AI-powered system that automates the extraction, validation, and organization of data from invoice documents. Built using LangChain, it combines large language models (LLMs) with specialized tools to read PDFs or images, extract fields like invoice numbers, dates, line items, and totals, cross-reference them against business rules, and output structured JSON or database records. Unlike a simple parser, an agent can reason about ambiguous layouts, handle missing data gracefully, and even flag discrepancies for human review.
Think of it as a digital accounts payable clerk that works 24/7, never makes typo errors, and scales effortlessly from 10 invoices a day to 10,000.
Why Automate Invoice Processing with LangChain?
Manual invoice processing is slow, expensive, and error-prone. Companies spend an average of $15–$30 per invoice when handled manually, and mistakes lead to payment delays, duplicate payments, and strained vendor relationships. Automating this workflow matters because:
- Cost Reduction: Cut processing costs by 70–90% compared to manual data entry.
- Speed: Process hundreds of invoices in minutes instead of days.
- Accuracy: LLMs with structured output parsing dramatically reduce extraction errors.
- Scalability: Handle growing invoice volumes without hiring additional staff.
- Auditability: Every extraction is logged and traceable for compliance.
LangChain is particularly well-suited because it provides a unified framework to chain document loaders, LLM calls, structured output parsers, and custom validation tools — all while maintaining context across steps. You get a composable, testable pipeline rather than a monolithic black box.
Architecture Overview
A well-designed invoice agent typically follows a multi-stage pipeline:
- Stage 1 — Ingestion: Load invoice PDFs or images using document loaders. Split long documents into manageable chunks while preserving table structures.
- Stage 2 — Extraction: Use an LLM with a structured output parser to pull key fields (invoice number, vendor name, date, line items, subtotal, tax, total) into a typed schema.
- Stage 3 — Verification: Run validation rules — does the sum of line items match the total? Is the date in a valid range? Are required fields present? Flag anomalies.
- Stage 4 — Summarization & Action: Generate a human-readable summary, store results in a database, or trigger downstream workflows like approval or payment.
The LangChain agent orchestrates these stages, deciding when to call which tool based on the current state of processing. This allows dynamic handling of edge cases — for example, if extraction confidence is low, the agent may re-read a specific chunk or escalate for human review.
Setting Up the Environment
Installing Dependencies
Start by installing the required packages. You'll need LangChain core, OpenAI integration, PDF loaders, and structured output utilities:
# Create a new project directory
mkdir invoice-agent && cd invoice-agent
# Create a virtual environment
python -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activate
# Install core dependencies
pip install langchain langchain-openai langchain-community
pip install pypdf pdfplumber # PDF parsing
pip install pillow pytesseract # OCR for image-based invoices
pip install pandas # For data manipulation
pip install python-dotenv # Environment variable managementSetting Up API Keys
Create a .env file to store your API keys securely. The agent will use OpenAI for LLM calls, but you can swap in other providers:
# .env file
OPENAI_API_KEY=sk-your-openai-api-key-here
# Optional: For using alternative models
# ANTHROPIC_API_KEY=sk-ant-your-keyLoad these in your application entry point:
# config.py
import os
from dotenv import load_dotenv
load_dotenv()
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
if not OPENAI_API_KEY:
raise ValueError("OPENAI_API_KEY not found in environment variables")Building the Core Components
Document Loading and Intelligent Splitting
Invoices can be multi-page PDFs with tables that span page breaks. A naive text split would destroy table context, making extraction unreliable. Use a PDF loader that preserves layout, then split with awareness of structural boundaries:
# document_loader.py
from langchain_community.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from typing import List
from langchain.schema import Document
def load_invoice_pdf(file_path: str) -> List[Document]:
"""
Load a PDF invoice and split it into semantically meaningful chunks.
Uses PyPDFLoader for initial loading and a custom splitter
that respects page boundaries and table structures.
"""
loader = PyPDFLoader(file_path)
raw_documents = loader.load()
# Use a splitter with larger chunk size to keep tables intact
# and significant overlap to preserve context across boundaries
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=2000,
chunk_overlap=400,
separators=["\n\n\n", "\n\n", "\n", " ", ""],
length_function=len,
)
split_docs = text_splitter.split_documents(raw_documents)
print(f"Loaded {len(raw_documents)} pages, split into {len(split_docs)} chunks")
return split_docs
# Example usage:
# docs = load_invoice_pdf("invoices/sample_invoice.pdf")For image-based invoices (scanned documents), add an OCR preprocessing step:
# ocr_loader.py
from PIL import Image
import pytesseract
from langchain.schema import Document
def ocr_invoice_image(image_path: str) -> Document:
"""
Extract text from a scanned invoice image using Tesseract OCR.
Returns a LangChain Document ready for the extraction pipeline.
"""
image = Image.open(image_path)
# Configure Tesseract for better extraction
# --psm 6: Assume a uniform block of text
# --oem 3: Use LSTM OCR engine
custom_config = r'--oem 3 --psm 6 -c tessedit_char_whitelist=0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ.,$€£%()-/@\n '
extracted_text = pytesseract.image_to_string(
image,
config=custom_config
)
return Document(
page_content=extracted_text,
metadata={"source": image_path, "type": "ocr_invoice"}
)
# Example usage:
# doc = ocr_invoice_image("invoices/scanned_invoice.jpg")
# print(doc.page_content[:500])Creating the Extraction Chain with Structured Output
This is the heart of the agent. Define a Pydantic schema for invoice data, then create a chain that prompts the LLM to extract structured information. Using LangChain's StructuredOutputParser ensures the output is always a valid object you can programmatically use:
# extraction_schema.py
from pydantic import BaseModel, Field
from typing import List, Optional
from datetime import date
from decimal import Decimal
class LineItem(BaseModel):
"""A single line item from an invoice."""
description: str = Field(description="Product or service description")
quantity: int = Field(description="Quantity of items", ge=0)
unit_price: Decimal = Field(description="Price per unit in the invoice currency")
line_total: Decimal = Field(description="Total for this line (quantity * unit_price)")
tax_rate: Optional[Decimal] = Field(default=None, description="Tax rate applied to this line")
class InvoiceData(BaseModel):
"""Complete structured data extracted from an invoice."""
invoice_number: str = Field(description="Unique invoice identifier, e.g., 'INV-2024-001'")
vendor_name: str = Field(description="Name of the company or person issuing the invoice")
vendor_address: Optional[str] = Field(default=None, description="Vendor's physical or billing address")
vendor_tax_id: Optional[str] = Field(default=None, description="VAT/Tax identification number of vendor")
customer_name: Optional[str] = Field(default=None, description="Name of the customer being billed")
invoice_date: date = Field(description="Date the invoice was issued, in YYYY-MM-DD format")
due_date: Optional[date] = Field(default=None, description="Payment due date, if specified")
currency: str = Field(default="USD", description="ISO 4217 currency code, e.g., USD, EUR, GBP")
line_items: List[LineItem] = Field(description="All line items on the invoice", min_items=1)
subtotal: Decimal = Field(description="Sum of all line totals before tax")
tax_total: Decimal = Field(description="Total tax amount")
shipping_handling: Optional[Decimal] = Field(default=None, description="Shipping or handling charges")
discount: Optional[Decimal] = Field(default=None, description="Any discounts applied")
grand_total: Decimal = Field(description="Final total to be paid (subtotal + tax + shipping - discount)")
payment_terms: Optional[str] = Field(default=None, description="Payment terms, e.g., 'Net 30'")
notes: Optional[str] = Field(default=None, description="Any additional notes or remarks on the invoice")Now build the extraction chain that takes raw document text and returns a populated InvoiceData object:
# extraction_chain.py
from langchain_openai import ChatOpenAI
from langchain.output_parsers import PydanticOutputParser
from langchain.prompts import ChatPromptTemplate, HumanMessagePromptTemplate
from langchain.schema import SystemMessage
from extraction_schema import InvoiceData
class InvoiceExtractor:
"""Extracts structured invoice data from document text using an LLM."""
def __init__(self, model_name: str = "gpt-4o-mini", temperature: float = 0):
self.llm = ChatOpenAI(
model=model_name,
temperature=temperature, # Zero temperature for consistent extraction
max_tokens=2000
)
self.parser = PydanticOutputParser(pydantic_object=InvoiceData)
# Build the prompt with format instructions baked in
self.system_prompt = SystemMessage(content=(
"You are an expert invoice data extraction assistant. "
"Your task is to read the provided invoice text and extract "
"all relevant fields into a structured JSON format.\n\n"
"CRITICAL INSTRUCTIONS:\n"
"- Extract ALL line items with their descriptions, quantities, prices, and totals.\n"
"- For monetary values, extract them as decimal numbers WITHOUT currency symbols.\n"
"- If a field is not present, leave it as null — do NOT hallucinate values.\n"
"- Verify that line totals sum to the subtotal, and subtotal + tax + shipping - discount equals grand_total.\n"
"- Dates should be in YYYY-MM-DD format.\n"
"- If you see handwritten or OCR text with possible errors, note it in the 'notes' field.\n"
"- Be precise about tax identification numbers and invoice numbers."
))
self.human_template = HumanMessagePromptTemplate.from_template(
"INVOICE DOCUMENT TEXT:\n{invoice_text}\n\n"
"{format_instructions}\n\n"
"Please extract the invoice data now."
)
self.prompt = ChatPromptTemplate.from_messages([
self.system_prompt,
self.human_template
])
# Wire up the chain: prompt -> LLM -> parser
self.chain = self.prompt | self.llm | self.parser
def extract(self, document_text: str) -> InvoiceData:
"""
Extract structured invoice data from raw text.
Returns a validated InvoiceData Pydantic object.
"""
format_instructions = self.parser.get_format_instructions()
result: InvoiceData = self.chain.invoke({
"invoice_text": document_text,
"format_instructions": format_instructions
})
return result
# Example usage:
# extractor = InvoiceExtractor()
# sample_text = open("sample_invoice_text.txt").read()
# invoice_data = extractor.extract(sample_text)
# print(invoice_data.model_dump_json(indent=2))Building the Verification Agent Tool
Extraction alone isn't enough. A verification step catches errors, validates business logic, and flags suspicious data. This tool becomes one of the agent's callable functions:
# verification_tool.py
from langchain.tools import tool
from extraction_schema import InvoiceData
from decimal import Decimal
from datetime import date, timedelta
from typing import List, Dict
@tool
def verify_invoice_data(invoice_json: str) -> str:
"""
Verify extracted invoice data for consistency and validity.
Input: A JSON string representation of InvoiceData.
Returns: A verification report with status, flags, and recommendations.
Checks performed:
- Mathematical consistency (line totals vs subtotal vs grand total)
- Date validity (invoice date not in the future, due date after invoice date)
- Required fields present
- Suspicious patterns (round numbers, missing tax, unusual discounts)
"""
import json
try:
data_dict = json.loads(invoice_json)
except json.JSONDecodeError:
return "ERROR: Invalid JSON input. Cannot verify."
flags: List[Dict[str, str]] = []
all_ok = True
# --- Check 1: Mathematical consistency ---
line_items = data_dict.get("line_items", [])
calculated_subtotal = Decimal("0")
for item in line_items:
qty = Decimal(str(item.get("quantity", 0)))
price = Decimal(str(item.get("unit_price", 0)))
line_total_from_item = Decimal(str(item.get("line_total", 0)))
expected_line_total = qty * price
if abs(line_total_from_item - expected_line_total) > Decimal("0.01"):
flags.append({
"type": "math_error",
"severity": "HIGH",
"detail": f"Line item '{item.get('description', 'Unknown')[:50]}' has total {line_total_from_item} "
f"but quantity * price = {expected_line_total}"
})
all_ok = False
calculated_subtotal += line_total_from_item
stated_subtotal = Decimal(str(data_dict.get("subtotal", 0)))
if abs(calculated_subtotal - stated_subtotal) > Decimal("0.02"):
flags.append({
"type": "subtotal_mismatch",
"severity": "HIGH",
"detail": f"Sum of line totals ({calculated_subtotal}) does not match stated subtotal ({stated_subtotal})"
})
all_ok = False
# Grand total reconciliation
tax = Decimal(str(data_dict.get("tax_total", 0)))
shipping = Decimal(str(data_dict.get("shipping_handling", 0)))
discount = Decimal(str(data_dict.get("discount", 0)))
expected_grand = stated_subtotal + tax + shipping - discount
stated_grand = Decimal(str(data_dict.get("grand_total", 0)))
if abs(expected_grand - stated_grand) > Decimal("0.05"):
flags.append({
"type": "grand_total_mismatch",
"severity": "CRITICAL",
"detail": f"Expected grand total {expected_grand} but stated as {stated_grand}. "
f"Difference: {expected_grand - stated_grand}"
})
all_ok = False
# --- Check 2: Date validity ---
today = date.today()
inv_date_str = data_dict.get("invoice_date")
if inv_date_str:
try:
inv_date = date.fromisoformat(inv_date_str)
if inv_date > today:
flags.append({
"type": "future_date",
"severity": "MEDIUM",
"detail": f"Invoice date {inv_date_str} is in the future"
})
all_ok = False
except ValueError:
flags.append({
"type": "invalid_date",
"severity": "HIGH",
"detail": f"Cannot parse invoice_date: {inv_date_str}"
})
all_ok = False
due_date_str = data_dict.get("due_date")
if inv_date_str and due_date_str:
try:
inv_date = date.fromisoformat(inv_date_str)
due_date = date.fromisoformat(due_date_str)
if due_date < inv_date:
flags.append({
"type": "due_before_invoice",
"severity": "HIGH",
"detail": f"Due date {due_date_str} is before invoice date {inv_date_str}"
})
all_ok = False
except ValueError:
pass # Already flagged above
# --- Check 3: Required fields ---
required_fields = ["invoice_number", "vendor_name", "grand_total"]
for field in required_fields:
val = data_dict.get(field)
if val is None or (isinstance(val, str) and val.strip() == ""):
flags.append({
"type": "missing_field",
"severity": "HIGH",
"detail": f"Required field '{field}' is missing or empty"
})
all_ok = False
# --- Check 4: Suspicious patterns ---
if data_dict.get("discount") and Decimal(str(data_dict["discount"])) > stated_subtotal * Decimal("0.3"):
flags.append({
"type": "large_discount",
"severity": "LOW",
"detail": "Discount exceeds 30% of subtotal — verify manually"
})
if len(line_items) > 20:
flags.append({
"type": "many_line_items",
"severity": "LOW",
"detail": f"Invoice has {len(line_items)} line items — review for duplicates"
})
# --- Build report ---
if all_ok and not flags:
return json.dumps({
"status": "PASSED",
"message": "All verification checks passed. Invoice data is consistent.",
"flags": []
}, indent=2)
else:
return json.dumps({
"status": "FAILED" if any(f["severity"] in ("HIGH", "CRITICAL") for f in flags) else "WARNING",
"message": f"Found {len(flags)} issue(s). Review required.",
"flags": flags
}, indent=2)
# The tool is decorated with @tool, making it directly usable in a LangChain agentBuilding the Summarization Tool
After extraction and verification, generate a concise human-readable summary suitable for dashboards, emails, or approval workflows:
# summarization_tool.py
from langchain.tools import tool
from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate
_summary_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3, max_tokens=500)
SUMMARY_PROMPT = PromptTemplate.from_template(
"""You are an accounts payable assistant. Create a concise, professional summary
of the following verified invoice data for management review.
INVOICE DATA (JSON):
{invoice_json}
VERIFICATION REPORT:
{verification_report}
Generate a 3-5 sentence summary covering:
1. Who the invoice is from and for how much
2. Key dates (issued, due)
3. Any flags or issues from verification
4. A recommended action (approve, reject, or review)
SUMMARY:"""
)
@tool
def summarize_invoice(invoice_json: str, verification_report: str) -> str:
"""
Generate a human-readable summary of an invoice with verification results.
Inputs:
- invoice_json: JSON string of the extracted InvoiceData
- verification_report: JSON string from the verification tool
Returns a plain-text summary suitable for email or dashboard display.
"""
chain = SUMMARY_PROMPT | _summary_llm
result = chain.invoke({
"invoice_json": invoice_json,
"verification_report": verification_report
})
return result.content
# Example:
# summary = summarize_invoice.invoke({
# "invoice_json": invoice_data.model_dump_json(),
# "verification_report": verification_result
# })
# print(summary)Assembling the LangChain Agent
Defining Tools
The agent needs access to all the tools you've built. Bundle them into a list that the agent can call dynamically:
# tools_registry.py
from verification_tool import verify_invoice_data
from summarization_tool import summarize_invoice
from langchain.tools import Tool
# Additional utility: a simple calculator for the agent to verify math
@tool
def calculate_line_total(quantity: int, unit_price: float) -> str:
"""Multiply quantity by unit price and return the result as a string."""
result = quantity * unit_price
return f"Line total: {result:.2f}"
# Additional utility: lookup known vendor database (simulated)
@tool
def lookup_vendor(vendor_name: str) -> str:
"""
Look up a vendor in the known vendor database.
Returns vendor details if found, or 'NOT_FOUND' if unknown.
"""
# Simulated database — in production, query a real DB
known_vendors = {
"acme corp": "TAX_ID: ACME123 | Payment Terms: Net 30 | Status: ACTIVE",
"globex industries": "TAX_ID: GBX456 | Payment Terms: Net 15 | Status: ACTIVE",
"initech": "TAX_ID: INT789 | Payment Terms: Net 45 | Status: ACTIVE",
}
vendor_lower = vendor_name.lower().strip()
if vendor_lower in known_vendors:
return known_vendors[vendor_lower]
return "NOT_FOUND"
# Master tool list for the agent
agent_tools = [
verify_invoice_data,
summarize_invoice,
calculate_line_total,
lookup_vendor,
]Creating the Agent Executor
Now wire everything into a LangChain agent. The agent receives a document, uses the extractor to get structured data, then iteratively calls verification, vendor lookup, and summarization tools as needed:
# invoice_agent.py
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from extraction_chain import InvoiceExtractor
from tools_registry import agent_tools
from document_loader import load_invoice_pdf
from typing import Dict, Any
class InvoiceProcessingAgent:
"""
A complete LangChain agent that orchestrates invoice processing:
load -> extract -> verify -> enrich -> summarize.
"""
def __init__(self, model_name: str = "gpt-4o"):
self.extractor = InvoiceExtractor()
# The agent uses a more capable model for reasoning about tool usage
self.agent_llm = ChatOpenAI(
model=model_name,
temperature=0,
max_tokens=2000
)
# Agent prompt with clear instructions on the workflow
self.agent_prompt = ChatPromptTemplate.from_messages([
("system",
"""You are an intelligent invoice processing agent. Your job is to process
invoices through a structured pipeline:
STEP 1: The user will provide extracted invoice data as JSON.
STEP 2: You MUST call 'verify_invoice_data' with the JSON to check for errors.
STEP 3: If verification flags issues, analyze them and note recommendations.
STEP 4: Call 'lookup_vendor' to enrich vendor information.
STEP 5: Call 'summarize_invoice' with the invoice JSON AND verification report.
STEP 6: Present the final summary to the user along with any recommended actions.
ALWAYS follow ALL steps in order. Do not skip verification.
If verification shows CRITICAL issues, recommend REJECTION.
If verification shows HIGH issues, recommend MANUAL REVIEW.
If verification passes, recommend APPROVAL.
Use the tools provided. Be thorough and professional."""),
("user", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad"),
])
# Create the OpenAI tools agent
self.agent = create_openai_tools_agent(
llm=self.agent_llm,
tools=agent_tools,
prompt=self.agent_prompt
)
self.executor = AgentExecutor(
agent=self.agent,
tools=agent_tools,
verbose=True,
max_iterations=10,
handle_parsing_errors=True,
return_intermediate_steps=True
)
def process_invoice(self, file_path: str) -> Dict[str, Any]:
"""
End-to-end processing of an invoice PDF.
Returns a dictionary with:
- extracted_data: The structured InvoiceData as a dict
- verification_report: Results from the verification step
- vendor_info: Enriched vendor data (if found)
- summary: Human-readable summary
- recommendation: APPROVE, REVIEW, or REJECT
- intermediate_steps: Full agent trace for auditing
"""
print(f"\n{'='*60}")
print(f"Processing invoice: {file_path}")
print(f"{'='*60}\n")
# Step 1: Load document
print("[Stage 1/4] Loading document...")
documents = load_invoice_pdf(file_path)
full_text = "\n\n".join([doc.page_content for doc in documents])
# Step 2: Extract structured data
print("[Stage 2/4] Extracting structured data...")
invoice_data = self.extractor.extract(full_text)
invoice_json = invoice_data.model_dump_json(indent=2)
print(f" Extracted invoice #{invoice_data.invoice_number} "
f"from {invoice_data.vendor_name}")
# Step 3-6: Agent handles verification, lookup, summarization
print("[Stage 3-6/4] Agent processing (verify, lookup, summarize)...")
agent_input = (
f"Here is the extracted invoice data as JSON:\n\n"
f"json\n{invoice_json}\n\n\n"
f"Please follow the pipeline: verify this data, look up the vendor, "
f"and provide a final summary with a recommendation."
)
result = self.executor.invoke({
"input": agent_input
})
return {
"extracted_data": invoice_data.model_dump(),
"agent_output": result.get("output", ""),
"intermediate_steps": [
{"action": step[0].tool, "input": step[0].tool_input, "output": step[1]}
for step in result.get("intermediate_steps", [])
],
"status": "COMPLETE"
}
# --- Run the agent ---
if __name__ == "__main__":
agent = InvoiceProcessingAgent()
result = agent.process_invoice("invoices/sample_invoice.pdf")
print("\n" + "="*60)
print("FINAL AGENT OUTPUT:")
print("="*60)
print(result["agent_output"])
print("\n" + "="*60)
print("EXTRACTED DATA (structured):")
print("="*60)
print(json.dumps(result["extracted_data"], indent=2, default=str))Putting It All Together — Complete Pipeline Script
Here is a self-contained script that runs the entire pipeline end-to-end. Save it as run_pipeline.py and execute it against a directory of invoices:
# run_pipeline.py
"""
Complete Invoice Processing Pipeline
Usage: python run_pipeline.py --input-dir ./invoices --output-dir ./results
"""
import argparse
import json
import os
from pathlib import Path
from datetime import datetime
from invoice_agent import InvoiceProcessingAgent
import pandas as pd
def process_directory(input_dir: str, output_dir: str):
"""Process all PDF invoices in a directory and save results."""
input_path = Path(input_dir)
output_path = Path(output_dir)
output_path.mkdir(parents=True, exist_ok=True)
pdf_files = list(input_path.glob("*.pdf"))
print(f"Found {len(pdf_files)} PDF files to process")
agent = InvoiceProcessingAgent()
all_results = []
for i, pdf_file in enumerate(pdf_files, 1):
print(f"\nProcessing file {i}/{len(pdf_files)}: {pdf_file.name}")
try:
result = agent.process_invoice(str(pdf_file))
# Save individual result
result_filename = pdf_file.stem + "_result.json"
result_path = output_path / result_filename
with open(result_path, "w") as f:
json.dump(result, f, indent=2, default=str)
# Collect summary for aggregate report
extracted = result.get("extracted_data", {})
all_results.append({
"filename": pdf_file.name,
"invoice_number": extracted.get("invoice_number", "UNKNOWN"),
"vendor_name": extracted.get("vendor_name", "UNKNOWN"),
"invoice_date": extracted.get("invoice_date"),
"grand_total": float(extracted.get("grand_total", 0)),
"currency": extracted.get("currency", "USD"),
"agent_summary": result.get("agent_output", "")[:200],
"status": result.get("status", "ERROR")
})
print(f" ✓ Saved result to {result_path}")
except Exception as e:
print(f" ✗ Error processing {pdf_file.name}: {str(e)}")
all_results.append({
"filename": pdf_file.name,
"status": f"ERROR: {str(e)[:100]}"
})
# Generate aggregate report
df = pd.DataFrame(all_results)
report_path = output_path / f"aggregate_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
df.to_csv(report_path, index=False)
print(f"\n{'='*60}")
print(f"Pipeline complete. Processed {len(pdf_files)} invoices.")
print(f"Aggregate report saved to: {report_path}")
print(f"Individual results in: {output_path}")
# Quick stats
success_count = sum(1 for r in all_results if r.get("status") == "COMPLETE")
total_value = sum(r.get("grand_total", 0) for r in all_results if r.get("grand_total"))
print(f"Successfully processed: {success_count}/{len(pdf_files)}")
print(f"Total invoice value: ${total_value:,.2f}")
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Process invoices with LangChain agent")
parser.add_argument("--input-dir", required=True, help="Directory containing invoice PDFs")
parser.add_argument("--output-dir", required=True, help="Directory for results")
args = parser.parse_args()
process_directory(args.input_dir, args.output_dir)Best Practices for Production Invoice Agents
When moving from prototype to production, keep these practices in mind:
- Confidence Scoring: Ask the LLM to return a confidence score (0–1) for each extracted field. Route low-confidence items to a human-in-the-loop queue. LangChain's
PydanticOutputParsercan be extended with an additional confidence field. - Idempotency and Caching: Store processed invoice hashes (SHA-256 of file content) in a database. If the same file is uploaded twice, return the cached result immediately instead of re-processing. This saves costs and prevents duplicate entries.
- Retry Logic with Exponential Backoff: LLM APIs occasionally fail. Wrap extraction calls in a retry decorator. LangChain integrates well with
tenacityfor this purpose. - Streaming for Large Batches: For high-volume processing, use LangChain's async capabilities (
ainvoke,abatch) to process multiple invoices concurrently while respecting rate limits. - Validation Layers, Not Just One: Combine LLM-based verification with deterministic rule checks (like the mathematical checks in our verification tool). LLMs are great at semantic understanding but can hallucinate on arithmetic — always back them up with code-based math validation.
- Human-in-the-Loop Escalation: Build a review interface for flagged invoices. The agent should produce a structured "review package" containing the original PDF, extracted data, flagged issues, and a suggested resolution — all ready for a human reviewer to approve or override in seconds.
- Monitor Costs and Token Usage: Use LangChain's callback system to log token counts per invoice. Set budget alerts. A typical invoice extraction might use 1,000–4,000 tokens depending on complexity. With
gpt-4o-mini, this costs fractions of a cent per invoice. - Version Your Schemas: Invoice formats evolve. Keep your
InvoiceDataPydantic model under version control with semantic versioning. Maintain backward compatibility so old extraction results remain valid. - Security Considerations: Invoice data is sensitive. Never log full invoice contents to plaintext. Mask vendor tax IDs and financial figures in logs. Use environment variables for all secrets, and consider running the agent within a VPC if processing highly confidential vendor data.