What is an Invoice Processing Agent with LangGraph?
An Invoice Processing Agent built with LangGraph is a stateful, multi-step AI workflow that automates the extraction, validation, and structuring of invoice data from unstructured inputs—such as scanned PDFs, images, or raw text. LangGraph, an orchestration framework from the LangChain ecosystem, lets you model this entire pipeline as a directed graph of nodes (processing steps) and edges (transitions), with built-in support for conditional branching, loops, and state management.
Unlike a single LLM call that tries to do everything in one shot, a LangGraph-based agent breaks the problem into discrete, composable stages: reading the document, extracting fields, validating line items, flagging discrepancies, enriching data from external sources, and finally producing a clean JSON or CSV output. Each node in the graph is a pure function that receives the current state and returns updates, making the system highly debuggable, testable, and extensible.
Why LangGraph for Invoice Processing?
Invoice processing presents several challenges that make LangGraph an ideal choice:
- Multi-step reasoning: Invoices require field extraction, cross-field validation (does the sum of line items match the total?), currency normalization, and sometimes supplier lookup—all tasks that benefit from separate, focused LLM calls.
- Conditional routing: Not all invoices are the same. Some need OCR, others come as plain text. Some have missing fields and need human-in-the-loop intervention. LangGraph's conditional edges let you dynamically route based on the state.
- State persistence: LangGraph maintains a typed state object across nodes, so intermediate results (raw OCR text, extracted entities, validation errors) are preserved and accessible at any stage.
- Human-in-the-loop: When the agent is uncertain, you can pause execution, surface the current state to a human reviewer via a UI, and resume with corrections—all within the same graph.
- Streaming and observability: LangGraph supports streaming node outputs, making it easy to build real-time dashboards that show progress through the pipeline.
Prerequisites and Setup
Before building the agent, install the required packages. This tutorial uses LangGraph, LangChain, and an LLM provider (OpenAI in the examples, but you can substitute any LangChain-compatible model).
pip install langgraph langchain langchain-openai pydantic python-dotenv
Set your OpenAI API key as an environment variable or via a .env file:
# .env
OPENAI_API_KEY=sk-your-key-here
Create a new Python file—let's call it invoice_agent.py—and import the necessary modules:
import os
from typing import TypedDict, List, Optional, Annotated, Literal
from dotenv import load_dotenv
from pydantic import BaseModel, Field
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage
load_dotenv()
Step-by-Step Implementation
Step 1: Define the State Schema
The state is the central data structure that flows through every node in the graph. For invoice processing, we need to track: the raw input, extracted structured data, validation errors, processing stage, and any human corrections. We use a TypedDict with Annotated reducers to control how updates merge.
class LineItem(BaseModel):
"""Schema for a single line item in an invoice."""
description: str = Field(description="Product or service description")
quantity: Optional[float] = Field(None, description="Quantity ordered")
unit_price: Optional[float] = Field(None, description="Price per unit")
line_total: Optional[float] = Field(None, description="Total for this line")
class InvoiceData(BaseModel):
"""Complete structured invoice data extracted by the agent."""
invoice_number: Optional[str] = Field(None, description="Invoice identifier")
invoice_date: Optional[str] = Field(None, description="Date of the invoice")
due_date: Optional[str] = Field(None, description="Payment due date")
supplier_name: Optional[str] = Field(None, description="Name of the supplier/vendor")
supplier_address: Optional[str] = Field(None, description="Supplier's address")
customer_name: Optional[str] = Field(None, description="Name of the customer/buyer")
subtotal: Optional[float] = Field(None, description="Sum before taxes and discounts")
tax_rate: Optional[float] = Field(None, description="Tax rate applied")
tax_amount: Optional[float] = Field(None, description="Total tax amount")
discount: Optional[float] = Field(None, description="Any discount applied")
total_amount: Optional[float] = Field(None, description="Final total due")
currency: Optional[str] = Field(None, description="Currency code (USD, EUR, etc.)")
line_items: List[LineItem] = Field(default_factory=list, description="All line items")
notes: Optional[str] = Field(None, description="Additional notes or payment instructions")
class InvoiceAgentState(TypedDict):
"""Complete state for the invoice processing agent."""
# Input
raw_text: str
raw_image_base64: Optional[str]
# Processing stages
extracted_data: Optional[dict]
validation_errors: List[str]
enriched_data: Optional[dict]
# Control flow
processing_stage: str # "ocr", "extraction", "validation", "enrichment", "done", "error"
needs_human_review: bool
human_corrections: Optional[dict]
# Final output
final_invoice: Optional[dict]
audit_log: List[str]
The InvoiceData Pydantic model serves as both the extraction target and the validation schema. The InvoiceAgentState dict holds all mutable state across the graph's lifecycle.
Step 2: Build the Graph Nodes
Each node is a pure function that takes the current state, performs its task, and returns a partial state update. We'll implement five core nodes: preprocess, extract, validate, enrich, and finalize.
Node 1: Preprocess / OCR
This node handles the initial input. If the invoice comes as a base64-encoded image, we simulate OCR (in production, you'd integrate a vision model like GPT-4V or a dedicated OCR service). If it's already raw text, we pass it through.
def preprocess_node(state: InvoiceAgentState) -> dict:
"""
Preprocess the input: perform OCR if needed, clean text,
and prepare for extraction.
"""
audit = state.get("audit_log", [])
audit.append("[PREPROCESS] Starting preprocessing")
raw_text = state.get("raw_text", "")
raw_image = state.get("raw_image_base64")
# If we have an image and no text, simulate OCR with a vision-capable LLM
if raw_image and not raw_text:
llm = ChatOpenAI(model="gpt-4o", temperature=0)
messages = [
HumanMessage(content=[
{"type": "text", "text": "Extract ALL text from this invoice image verbatim. Return only the raw text, preserving the layout as much as possible."},
{"type": "image_url", "image_url": {"url": f"data:image/png;base64,{raw_image}"}}
])
]
response = llm.invoke(messages)
raw_text = response.content
audit.append("[PREPROCESS] OCR completed via vision model")
# Basic text cleaning
cleaned_text = raw_text.strip()
return {
"raw_text": cleaned_text,
"processing_stage": "extraction",
"audit_log": audit
}
Node 2: Extract Structured Data
This is the core extraction node. We feed the cleaned text to an LLM with a structured output prompt and use LangChain's with_structured_output to get a typed InvoiceData object directly. This eliminates manual parsing and ensures type safety.
def extract_node(state: InvoiceAgentState) -> dict:
"""
Extract structured invoice data from raw text using an LLM
with structured output parsing.
"""
audit = state.get("audit_log", [])
audit.append("[EXTRACT] Starting structured extraction")
llm = ChatOpenAI(model="gpt-4o", temperature=0)
structured_llm = llm.with_structured_output(InvoiceData)
raw_text = state.get("raw_text", "")
system_prompt = """You are an expert invoice data extraction system.
Extract the following fields from the invoice text below.
- If a field is not present, leave it as null/None.
- For line items, extract every row as a separate LineItem.
- Be precise with numeric values: do not include currency symbols in numeric fields.
- Dates should be in ISO format (YYYY-MM-DD) when possible.
Invoice text:
"""
messages = [HumanMessage(content=system_prompt + raw_text)]
result: InvoiceData = structured_llm.invoke(messages)
extracted_dict = result.model_dump(exclude_none=False)
audit.append(f"[EXTRACT] Extracted invoice #{extracted_dict.get('invoice_number', 'N/A')}")
return {
"extracted_data": extracted_dict,
"processing_stage": "validation",
"audit_log": audit
}
Node 3: Validate Extracted Data
Validation checks for internal consistency—does the sum of line totals (minus discounts plus tax) equal the stated total? Are required fields present? This node flags issues and decides whether human review is needed.
def validate_node(state: InvoiceAgentState) -> dict:
"""
Validate extracted invoice data for consistency and completeness.
Flags discrepancies and sets needs_human_review if critical issues found.
"""
audit = state.get("audit_log", [])
audit.append("[VALIDATE] Starting validation checks")
extracted = state.get("extracted_data", {})
errors = []
critical_issues = False
# Check required fields
required_fields = ["invoice_number", "invoice_date", "total_amount", "supplier_name"]
for field in required_fields:
if not extracted.get(field):
errors.append(f"MISSING_REQUIRED: {field} is empty or missing")
critical_issues = True
# Validate line item totals against subtotal
line_items = extracted.get("line_items", [])
if line_items:
computed_subtotal = sum(
item.get("line_total", 0) or 0
for item in line_items
)
stated_subtotal = extracted.get("subtotal")
if stated_subtotal and abs(computed_subtotal - stated_subtotal) > 0.01:
errors.append(
f"SUBTOTAL_MISMATCH: Sum of line totals ({computed_subtotal:.2f}) "
f"does not match stated subtotal ({stated_subtotal:.2f})"
)
critical_issues = True
# Validate total = subtotal - discount + tax
subtotal = extracted.get("subtotal") or 0
discount = extracted.get("discount") or 0
tax_amount = extracted.get("tax_amount") or 0
stated_total = extracted.get("total_amount")
if stated_total:
computed_total = subtotal - discount + tax_amount
if abs(computed_total - stated_total) > 0.01:
errors.append(
f"TOTAL_MISMATCH: Computed total ({computed_total:.2f}) "
f"does not match stated total ({stated_total:.2f})"
)
critical_issues = True
audit.append(f"[VALIDATE] Found {len(errors)} validation issues")
return {
"validation_errors": errors,
"needs_human_review": critical_issues,
"processing_stage": "human_review" if critical_issues else "enrichment",
"audit_log": audit
}
Node 4: Human Review Handler
When needs_human_review is True, the graph routes to this node. In a real application, this would pause execution and surface the state to a UI. For this tutorial, we simulate receiving corrections through the state's human_corrections field. The node merges corrections into the extracted data and re-runs validation.
def human_review_node(state: InvoiceAgentState) -> dict:
"""
Handle human-in-the-loop review. Applies corrections from
the human_corrections field and transitions back to validation.
"""
audit = state.get("audit_log", [])
audit.append("[HUMAN_REVIEW] Applying human corrections")
extracted = state.get("extracted_data", {})
corrections = state.get("human_corrections", {})
if corrections:
# Merge corrections into extracted data
for key, value in corrections.items():
if key in extracted:
extracted[key] = value
audit.append(f"[HUMAN_REVIEW] Corrected {key} to {value}")
return {
"extracted_data": extracted,
"human_corrections": {}, # Clear corrections after applying
"needs_human_review": False,
"processing_stage": "validation", # Re-validate after corrections
"audit_log": audit
}
Node 5: Enrich Data
Enrichment adds context—looking up supplier details in a database, normalizing currency codes, or fetching exchange rates. Here we simulate a supplier lookup and add a confidence score.
def enrich_node(state: InvoiceAgentState) -> dict:
"""
Enrich the validated invoice data with external lookups
and additional metadata.
"""
audit = state.get("audit_log", [])
audit.append("[ENRICH] Starting data enrichment")
extracted = state.get("extracted_data", {})
# Simulate supplier database lookup
supplier_db = {
"Acme Corp": {"id": "SUP-001", "category": "Industrial Supplies", "credit_rating": "A"},
"GlobalTech Ltd": {"id": "SUP-002", "category": "Technology", "credit_rating": "A+"},
}
supplier_name = extracted.get("supplier_name", "")
supplier_info = supplier_db.get(supplier_name, {})
enriched = dict(extracted)
enriched["supplier_metadata"] = supplier_info
enriched["extraction_confidence"] = 0.95 if not state.get("validation_errors") else 0.7
enriched["processed_at"] = "2025-01-15T10:30:00Z" # Use actual timestamp in production
audit.append(f"[ENRICH] Enriched with supplier metadata for '{supplier_name}'")
return {
"extracted_data": enriched,
"enriched_data": enriched,
"processing_stage": "finalize",
"audit_log": audit
}
Node 6: Finalize and Format Output
The final node produces the output in a clean format and marks processing as complete.
def finalize_node(state: InvoiceAgentState) -> dict:
"""
Finalize processing: format the output, log completion,
and set the stage to done.
"""
audit = state.get("audit_log", [])
enriched = state.get("enriched_data") or state.get("extracted_data", {})
# Build the final structured output
final = {
"invoice_number": enriched.get("invoice_number"),
"invoice_date": enriched.get("invoice_date"),
"due_date": enriched.get("due_date"),
"supplier": {
"name": enriched.get("supplier_name"),
"address": enriched.get("supplier_address"),
"metadata": enriched.get("supplier_metadata", {})
},
"customer_name": enriched.get("customer_name"),
"line_items": enriched.get("line_items", []),
"financial_summary": {
"subtotal": enriched.get("subtotal"),
"discount": enriched.get("discount"),
"tax_rate": enriched.get("tax_rate"),
"tax_amount": enriched.get("tax_amount"),
"total_amount": enriched.get("total_amount"),
"currency": enriched.get("currency", "USD")
},
"validation_errors": state.get("validation_errors", []),
"confidence": enriched.get("extraction_confidence", 1.0),
"notes": enriched.get("notes"),
"processed_at": enriched.get("processed_at")
}
audit.append("[FINALIZE] Processing complete")
return {
"final_invoice": final,
"processing_stage": "done",
"audit_log": audit
}
Step 3: Connect the Edges and Conditional Logic
Now we wire everything together in a StateGraph. The magic of LangGraph is in its conditional edges—they let the graph dynamically decide which node to execute next based on the current state.
def build_invoice_agent_graph():
"""
Construct the LangGraph state graph for invoice processing.
Returns a compiled graph ready for invocation.
"""
# Initialize the graph with our state schema
workflow = StateGraph(InvoiceAgentState)
# Add all nodes
workflow.add_node("preprocess", preprocess_node)
workflow.add_node("extract", extract_node)
workflow.add_node("validate", validate_node)
workflow.add_node("human_review", human_review_node)
workflow.add_node("enrich", enrich_node)
workflow.add_node("finalize", finalize_node)
# Set the entry point: always start with preprocessing
workflow.set_entry_point("preprocess")
# Simple edge: preprocess -> extract
workflow.add_edge("preprocess", "extract")
# Simple edge: extract -> validate
workflow.add_edge("extract", "validate")
# Conditional edge after validation:
# - If needs_human_review -> go to human_review node
# - If validation passed -> go to enrichment
workflow.add_conditional_edges(
"validate",
lambda state: "human_review" if state.get("needs_human_review") else "enrich",
{
"human_review": "human_review",
"enrich": "enrich"
}
)
# After human review, go back to validation (loop until clean)
workflow.add_edge("human_review", "validate")
# After enrichment, go to finalize
workflow.add_edge("enrich", "finalize")
# Finalize ends the graph
workflow.add_edge("finalize", END)
return workflow
The conditional edge after validate is the key decision point. If the validation node sets needs_human_review=True, the graph routes to human_review, which applies corrections and loops back to validate. This loop continues until all critical issues are resolved (in practice, you'd add a maximum loop counter to prevent infinite cycles). Once validation passes, it proceeds to enrichment and finalization.
Step 4: Compile and Execute the Agent
With the graph defined, we compile it—optionally adding a checkpointer for state persistence—and invoke it with sample invoice data.
def run_invoice_agent(raw_invoice_text: str, human_corrections: Optional[dict] = None):
"""
Compile and execute the invoice processing agent.
Returns the final processed invoice and audit log.
"""
# Build the graph
workflow = build_invoice_agent_graph()
# Add a memory checkpointer for state persistence between steps
memory = MemorySaver()
compiled_graph = workflow.compile(checkpointer=memory)
# Prepare the initial state
initial_state: InvoiceAgentState = {
"raw_text": raw_invoice_text,
"raw_image_base64": None,
"extracted_data": None,
"validation_errors": [],
"enriched_data": None,
"processing_stage": "ocr",
"needs_human_review": False,
"human_corrections": human_corrections or {},
"final_invoice": None,
"audit_log": []
}
# Execute the graph
# We use a thread_id for the checkpointer to track this session
config = {"configurable": {"thread_id": "invoice-session-001"}}
final_state = compiled_graph.invoke(
initial_state,
config
)
return final_state
Let's test the agent with a realistic invoice sample:
# Sample invoice text
sample_invoice = """
INVOICE
Invoice Number: INV-2025-0042
Invoice Date: 2025-01-12
Due Date: 2025-02-11
Supplier:
Acme Corp
123 Industrial Blvd, Chicago, IL 60601
Customer:
Beta Manufacturing Inc.
456 Enterprise Ave, Detroit, MI 48201
Line Items:
| Description | Qty | Unit Price | Line Total |
|----------------------|------|------------|------------|
| Steel Rods - Grade A | 100 | 12.50 | 1250.00 |
| Brass Fittings | 250 | 8.75 | 2187.50 |
| Assembly Labor | 40 | 35.00 | 1400.00 |
Subtotal: $4,837.50
Discount: $100.00
Tax (8.5%): $402.69
Total Amount Due: $5,140.19
Payment Terms: Net 30
Currency: USD
Notes: Please include PO number on all correspondence.
"""
# Run the agent
if __name__ == "__main__":
result = run_invoice_agent(sample_invoice)
print("=== FINAL INVOICE OUTPUT ===")
print(json.dumps(result["final_invoice"], indent=2, default=str))
print("\n=== AUDIT LOG ===")
for entry in result["audit_log"]:
print(f" {entry}")
print(f"\n=== PROCESSING STAGE: {result['processing_stage']} ===")
Expected output will show a clean JSON structure with all fields populated, the audit trail showing each step, and the final processing stage as "done".
Step 5: Adding Validation Error Handling and Human-in-the-Loop
To demonstrate the human review loop, let's invoke the agent with an invoice that has intentional discrepancies and provide corrections:
# Invoice with a deliberate mismatch
faulty_invoice = """
INVOICE #INV-2025-0099
Date: 2025-01-20
Supplier: GlobalTech Ltd, 789 Tech Park, San Francisco, CA 94107
Customer: Delta Systems
Items:
- Server Racks: 5 x $200.00 = $1000.00
- Network Cables: 100 x $15.00 = $1500.00
Subtotal: $2,500.00
Tax (10%): $250.00
Total: $2,800.00 <-- Mismatch: should be $2,750.00
"""
# First run - agent will flag the mismatch and pause for human review
result1 = run_invoice_agent(faulty_invoice)
print("Stage after first run:", result1["processing_stage"])
print("Needs human review:", result1["needs_human_review"])
print("Validation errors:", result1["validation_errors"])
# Now simulate human providing corrections via a second invocation
# In production, you'd use the checkpointer to resume the same thread
corrections = {
"total_amount": 2750.00, # Human confirms the correct total
}
# Re-run with corrections (in a real app, you'd resume the graph, not restart)
result2 = run_invoice_agent(faulty_invoice, human_corrections=corrections)
print("\nFinal invoice after corrections:")
print(json.dumps(result2["final_invoice"], indent=2, default=str))
In a production deployment, rather than restarting the graph, you would use LangGraph's graph.interrupt() to pause execution at the human_review node and later resume it with the same thread_id and updated state. Here's the pattern:
# Inside human_review_node in production:
def human_review_node_production(state: InvoiceAgentState) -> dict:
"""
Production version that interrupts for actual human input.
"""
from langgraph.errors import GraphInterrupt
if not state.get("human_corrections"):
# Pause and wait for human input
raise GraphInterrupt(
message="Please review and correct the flagged invoice data.",
data={
"extracted_data": state.get("extracted_data"),
"validation_errors": state.get("validation_errors")
}
)
# If corrections provided, apply them
extracted = state.get("extracted_data", {})
corrections = state.get("human_corrections", {})
for key, value in corrections.items():
extracted[key] = value
return {
"extracted_data": extracted,
"needs_human_review": False,
"processing_stage": "validation",
"audit_log": state.get("audit_log", []) + ["[HUMAN_REVIEW] Corrections applied"]
}
To resume a paused graph, you call compiled_graph.invoke with the same thread_id and pass None as the state (the checkpointer restores it), while providing the human corrections via a separate mechanism (like an external database update that the node reads).
Best Practices for Invoice Processing Agents
1. Use Structured Output Parsing
Always use with_structured_output (or equivalent) rather than parsing free-form LLM responses with regex. It dramatically reduces extraction errors and ensures type consistency. Define your Pydantic schemas with clear Field descriptions—these act as prompts to the LLM.
2. Implement Idempotent Nodes
Design each node function to be idempotent: given the same input state, it should produce the same output. This makes retries safe and debugging predictable. Avoid side effects inside node functions; instead, collect them in the audit log and apply them in a dedicated side-effect node if needed.
3. Add Comprehensive Validation
Go beyond simple field presence checks. Validate:
- Date consistency (due_date >= invoice_date)
- Tax rate plausibility (within expected ranges for the detected country)
- Line item totals match quantity × unit_price within rounding tolerance
- Currency code is valid ISO 4217
- Invoice number format matches known patterns for the supplier
4. Use Checkpointing for Resilience
Always attach a MemorySaver or persistent checkpointer (like SqliteSaver) to your compiled graph. This lets you resume failed executions, implement human-in-the-loop, and debug by inspecting state at any point in the graph's history.
5. Keep Nodes Small and Focused
Each node should do exactly one thing. Resist the temptation to combine extraction and validation into one node. Small nodes are easier to test, easier to swap out (e.g., replace the OCR node with a different provider), and allow finer-grained conditional routing.
6. Log Everything
Maintain an audit_log in your state that records every decision, extraction, and correction. This creates a complete trail useful for compliance, debugging, and building trust with users who can see exactly what the agent did.
7. Handle Edge Cases Gracefully
Build nodes for common edge cases:
- Multi-page invoices: A preprocess node that detects page breaks and concatenates text.
- Multi-currency: An enrichment node that fetches exchange rates and normalizes to a base currency.
- Handwritten invoices: Route to a specialized vision model node before extraction.
- Duplicate detection: A node that checks the extracted invoice number against a database of previously processed invoices.
8. Parallelize Where Possible
LangGraph supports branching to multiple nodes simultaneously. If you have independent enrichment tasks (supplier lookup, currency conversion, fraud check), run them in parallel branches that merge before finalization. This reduces end-to-end latency significantly.
# Example of parallel enrichment branches
workflow.add_conditional_edges(
"validate",
lambda state: "parallel_enrich", # routes to multiple nodes
{
"supplier_lookup": "supplier_lookup_node",
"currency_normalize": "currency_normalize_node",
"fraud_check": "fraud_check_node"
}
)
# Then merge results back into a single enrich node
workflow.add_edge("supplier_lookup_node", "merge_enrich")
workflow.add_edge("currency_normalize_node", "merge_enrich")
workflow.add_edge("fraud_check_node", "merge_enrich")
Conclusion
Building an invoice processing agent with LangGraph transforms a traditionally manual, error-prone workflow into a robust, auditable, and scalable AI pipeline. By decomposing the problem into discrete graph nodes—preprocessing, extraction, validation, human review, enrichment, and finalization—you gain fine-grained control over every step, the ability to pause for human intervention when needed, and a clear audit trail for every processed invoice.
The state-driven architecture means your agent remembers context across steps, handles edge cases through conditional routing, and can be extended with new nodes as business requirements evolve. Whether you're processing hundreds of supplier invoices daily or building a multi-tenant SaaS for accounts payable automation, LangGraph provides the orchestration layer that keeps your LLM-based extraction reliable, observable, and production-ready. Start with the skeleton in this tutorial, add your own business logic and integrations, and you'll have a powerful invoice processing system that improves with every iteration.