Document Processing Automation: Architecture Blueprint & Production Stack

from typing import TypedDict, Optional
from pydantic import BaseModel, Field
from langgraph.graph import StateGraph, END
import httpx

class InvoiceItem(BaseModel):
    description: str
    quantity: int = Field(gt=0)
    unit_price: float = Field(gt=0.0)
    total_amount: float

class InvoiceSchema(BaseModel):
    invoice_number: str
    vendor_name: str
    tax_id: str
    invoice_date: str
    items: list[InvoiceItem]
    grand_total: float

class DocumentState(TypedDict):
    image_bytes: bytes
    raw_response: Optional[str]
    parsed_invoice: Optional[InvoiceSchema]
    retry_count: int
    error_log: Optional[str]

async def vision_extraction_node(state: DocumentState) -> DocumentState:
    """Invokes vLLM Hosted Qwen2-VL endpoint with structured JSON mode."""
    async with httpx.AsyncClient() as client:
        res = await client.post(
            "http://vllm-inference.internal:8000/v1/chat/completions",
            json={
                "model": "Qwen2-VL-7B-Instruct",
                "messages": [{
                    "role": "user",
                    "content": [
                        {"type": "text", "text": "Extract all fields matching this schema accurately."},
                        {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{state['image_bytes']}"}}
                    ]
                }],
                "response_format": {"type": "json_object"}
            },
            timeout=10.0
        )
        data = res.json()
        state["raw_response"] = data["choices"][0]["message"]["content"]
        return state

def pydantic_validation_node(state: DocumentState) -> DocumentState:
    """Validates raw model output against the InvoiceSchema Pydantic model."""
    try:
        validated = InvoiceSchema.model_validate_json(state["raw_response"])
        state["parsed_invoice"] = validated
        state["error_log"] = None
    except Exception as err:
        state["retry_count"] += 1
        state["error_log"] = str(err)
    return state

def route_validation(state: DocumentState) -> str:
    if state.get("parsed_invoice") is not None:
        return "store_in_pgvector"
    if state["retry_count"] >= 2:
        return "human_escalation"
    return "reprompt_vision_node"

# LangGraph Workflow Definition
workflow = StateGraph(DocumentState)
workflow.add_node("vision_extract", vision_extraction_node)
workflow.add_node("validate_schema", pydantic_validation_node)
workflow.set_entry_point("vision_extract")
workflow.add_edge("vision_extract", "validate_schema")
workflow.add_conditional_edges("validate_schema", route_validation)
app = workflow.compile()